03 - Tool Context
Access runtime features through Context including logging, secrets, and progress reporting.
Running the Example
- Run:
uv run 03_context.py - Run (stdio):
uv run 03_context.py stdio - Env: set
API_KEY,DATABASE_URL
Source Code
#!/usr/bin/env python3
"""
03_context.py - Using Context with namespaced runtime APIs
This example shows how tools can access runtime features through
Context (provided at runtime by the TDK wrapper), including logging,
secrets, and progress reporting.
To run:
uv run 03_context.py # HTTP transport (default)
uv run 03_context.py stdio # stdio transport for Claude Desktop
Set environment variables for secrets:
export API_KEY="your-secret-key"
export DATABASE_URL="postgresql://localhost/mydb"
"""
import sys
from typing import Annotated, Any
from arcade_mcp_server import Context, MCPApp
# Create the MCP application
app = MCPApp(
name="context_example",
version="1.0.0",
instructions="Example server demonstrating Context usage",
)
@app.tool
async def secure_api_call(
context: Context,
endpoint: Annotated[str, "API endpoint to call"],
method: Annotated[str, "HTTP method (GET, POST, etc.)"] = "GET",
) -> Annotated[str, "API response or error message"]:
"""Make a secure API call using secrets from context."""
# Access secrets from environment via Context helper
try:
api_key = context.get_secret("API_KEY")
except ValueError:
await context.log.error("API_KEY not found in environment")
return "Error: API_KEY not configured"
# Log the API call
await context.log.info(f"Making {method} request to {endpoint}")
# Simulate API call (in real code, use httpx or aiohttp)
return f"Successfully called {endpoint} with API key: {api_key[:4]}..."
# Don't forget to add the secret to the .env file or export it as an environment variable
@app.tool(requires_secrets=["DATABASE_URL"])
async def database_info(
context: Context, table_name: Annotated[str | None, "Specific table to check"] = None
) -> Annotated[str, "Database connection info"]:
"""Get database connection information from context."""
# Get database URL from secrets
try:
db_url = context.get_secret("DATABASE_URL")
except ValueError:
db_url = "Not configured"
# Log at different levels
if db_url == "Not configured":
await context.log.warning("DATABASE_URL not set")
else:
await context.log.debug(f"Checking database: {db_url.split('@')[-1]}")
# Get user info
user_info = f"User: {context.user_id or 'anonymous'}"
if table_name:
return f"{user_info}\nDatabase: {db_url}\nChecking table: {table_name}"
else:
return f"{user_info}\nDatabase: {db_url}"
@app.tool
async def debug_context(
context: Context,
show_secrets: Annotated[bool, "Whether to show secret keys (not values)"] = False,
) -> Annotated[dict, "Current context information"]:
"""Debug tool to inspect the current context."""
info: dict[str, Any] = {
"user_id": context.user_id,
}
if show_secrets:
# Only show keys, not values for security
info["secret_keys"] = [s.key for s in (context.secrets or [])]
# Log that debug info was accessed
await context.log.info(f"Debug context accessed by {context.user_id or 'unknown'}")
return info
@app.tool
async def process_with_progress(
context: Context,
items: Annotated[list[str], "Items to process"],
delay_seconds: Annotated[float, "Delay between items"] = 0.1,
) -> Annotated[dict, "Processing results"]:
"""Process items with progress notifications."""
results: dict[str, list] = {"processed": [], "errors": []}
# Log start
await context.log.info(f"Starting to process {len(items)} items")
for i, item in enumerate(items):
try:
# Simulate processing
import asyncio
await asyncio.sleep(delay_seconds)
# Report progress (current, total, message)
await context.progress.report(i + 1, len(items), f"Processing: {item}")
await context.log.debug(f"Processing item {i + 1}/{len(items)}: {item}")
results["processed"].append(item.upper())
except Exception as e:
await context.log.error(f"Failed to process {item}: {e}")
results["errors"].append({"item": item, "error": str(e)})
# Log completion
await context.log.info(
f"Processing complete: {len(results['processed'])} succeeded, "
f"{len(results['errors'])} failed"
)
return results
# The Context provides at runtime (via TDK wrapper):
# - context.user_id: ID of the user making the request
# - context.get_secret(key): Retrieve a secret value (raises if missing)
# - context.log.<level>(msg): Send log messages to the client (debug/info/warning/error)
# - context.progress.report(progress, total=None, message=None): Progress updates
if __name__ == "__main__":
# Check if stdio transport was requested
transport = "stdio" if len(sys.argv) > 1 and sys.argv[1] == "stdio" else "http"
print(f"Starting {app.name} v{app.version}")
print(f"Transport: {transport}")
# Run the server
app.run(transport=transport, host="127.0.0.1", port=8000)
Context Features
The Context provides access to runtime features:
1. Logging
Send log messages at different levels:
await context.log.debug("Debug message")
await context.log.info("Information message")
await context.log.warning("Warning message")
await context.log.error("Error message")
2. Secrets Management
Access environment variables securely:
3. User Context
Access information about the current user:
4. Progress Reporting
Report progress for long-running operations:
5. Tool Decorator Options
Specify required secrets:
Key Concepts
- Context Parameter: Tools receive a
Contextas their first parameter - Async Functions: Use
async deffor tools that use context features - Secure Secrets: Secrets are accessed through context, not hardcoded
- Structured Logging: Log at appropriate levels for debugging
- Progress Updates: Keep users informed during long operations