Context is a class in Arcade that manages authorization and user information for requiring authentication. Context provides access to the following runtime features:
Logging via context.log
Secrets via context.get_secret
User via context.user_id—this is the user ID of the user invoking the
Progress reporting via context.progress.report
Decorator options
Some authorization providers may also include additional structured
information in the ToolContext.
How Context Works
When you invoke a that requires authorization, Arcade’s Tool SDK automatically:
Creates a Context object
Passes this object to your
You can then use the Context object to make authenticated requests to external APIs.
Let’s walk through an example (or skip to the full code).
Context Features
Context provides access to runtime features:
Logging
Logging in MCP provides a standardized channel for servers to emit structured, leveled messages to clients. Logging is useful for observability, debugging, and user feedback during .
You can send log messages at different levels using the log attribute of the Context object like this:
For the code to work, you must define your environment variables in a .env file.
Python
server.py
#!/usr/bin/env python3import sysfrom typing import Annotated, Anyfrom arcade_mcp_server import Context, MCPApp# Create the MCP applicationapp = MCPApp( name="context_example", version="1.0.0", instructions="Example server demonstrating Context usage",)@app.toolasync 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.toolasync 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.toolasync 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:# - 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 updatesif __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)
Run your MCP server
HTTP transport (default)stdio transport (for Claude Desktop)