# Existing FastMCP server becomes ETDI-secured with decorator from mcp.server.fastmcp import FastMCP app = FastMCP("My Server") # Standard tool (no security) @app.tool() def standard_tool(data: str) -> str: return f"Processed: {data}" # ETDI-secured tool with OAuth + Request Signing @app.tool(  etdi=True,  etdi_permissions=["data:read", "data:write"],  etdi_oauth_scopes=["tools:execute"],  etdi_require_request_signing=True ) def secure_tool(sensitive_data: str) -> str: return f"Securely processed: {sensitive_data}"

# ETDI client discovers ALL MCP servers (ETDI and non-ETDI) from mcp.etdi.client import ETDIClient client = ETDIClient(config) await client.connect_to_server(["python", "-m", "any_mcp_server"], "server-name") tools = await client.discover_tools() # Works with any MCP server!

# Add lifespan support for startup/shutdown with strong typing from contextlib import asynccontextmanager from collections.abc import AsyncIterator from dataclasses import dataclass from fake_database import Database # Replace with your actual DB type from mcp.server.fastmcp import FastMCP # Create a named server mcp = FastMCP("My App") # Specify dependencies for deployment and development mcp = FastMCP("My App", dependencies=["pandas", "numpy"]) @dataclass class AppContext: db: Database @asynccontextmanager async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """Manage application lifecycle with type-safe context""" # Initialize on startup db = await Database.connect() try: yield AppContext(db=db) finally: # Cleanup on shutdown await db.disconnect() # Pass lifespan to server mcp = FastMCP("My App", lifespan=app_lifespan) # Access type-safe lifespan context in tools @mcp.tool() def query_db() -> str: """Tool that uses initialized resources""" ctx = mcp.get_context() db = ctx.request_context.lifespan_context["db"] return db.query()

import asyncio from mcp.etdi import ETDIClient, OAuthConfig, SecurityLevel async def main(): # Configure OAuth provider oauth_config = OAuthConfig( provider="auth0", client_id="your-client-id", client_secret="your-client-secret", domain="your-domain.auth0.com", audience="https://your-api.example.com", scopes=["read:tools", "execute:tools"] ) # Initialize ETDI client async with ETDIClient({ "security_level": SecurityLevel.ENHANCED, "oauth_config": oauth_config.to_dict(), "allow_non_etdi_tools": True, "show_unverified_tools": False }) as client: # Connect to MCP servers await client.connect_to_server(["python", "-m", "my_server"], "my-server") # Discover and verify tools tools = await client.discover_tools() for tool in tools: if tool.verification_status.value == "verified": # Approve tool for usage await client.approve_tool(tool) # Invoke tool result = await client.invoke_tool(tool.id, {"param": "value"}) print(f"Result: {result}") asyncio.run(main())

import asyncio from mcp.etdi.server import ETDISecureServer from mcp.etdi import OAuthConfig async def main(): # Configure OAuth oauth_configs = [ OAuthConfig( provider="auth0", client_id="your-client-id", client_secret="your-client-secret", domain="your-domain.auth0.com", audience="https://your-api.example.com", scopes=["read:tools", "execute:tools"] ) ] # Create secure server server = ETDISecureServer(oauth_configs) # Register secure tool @server.secure_tool(permissions=["read:data", "write:data"]) async def secure_calculator(operation: str, a: float, b: float) -> float: """A secure calculator with OAuth protection""" if operation == "add": return a + b elif operation == "multiply": return a * b else: raise ValueError(f"Unknown operation: {operation}") await server.initialize() print("Secure server running with OAuth protection") asyncio.run(main())

from mcp.etdi import OAuthConfig auth0_config = OAuthConfig( provider="auth0", client_id="your-auth0-client-id", client_secret="your-auth0-client-secret", domain="your-domain.auth0.com", audience="https://your-api.example.com", scopes=["read:tools", "execute:tools"] )

okta_config = OAuthConfig( provider="okta", client_id="your-okta-client-id", client_secret="your-okta-client-secret", domain="your-domain.okta.com", scopes=["etdi.tools.read", "etdi.tools.execute"] )

from mcp.server.fastmcp import FastMCP, Context mcp = FastMCP("My App") @mcp.tool() async def long_task(files: list[str], ctx: Context) -> str: """Process multiple files with progress tracking""" for i, file in enumerate(files): ctx.info(f"Processing {file}") await ctx.report_progress(i, len(files)) data, mime_type = await ctx.read_resource(f"file://{file}") return "Processing complete"

from mcp import FastMCP from mcp.server.auth.provider import OAuthAuthorizationServerProvider from mcp.server.auth.settings import ( AuthSettings, ClientRegistrationOptions, RevocationOptions, ) class MyOAuthServerProvider(OAuthAuthorizationServerProvider): # See an example on how to implement at `examples/servers/simple-auth` ... mcp = FastMCP( "My App", auth_server_provider=MyOAuthServerProvider(), auth=AuthSettings( issuer_url="https://myapp.com", revocation_options=RevocationOptions( enabled=True, ), client_registration_options=ClientRegistrationOptions( enabled=True, valid_scopes=["myscope", "myotherscope"], default_scopes=["myscope"], ), required_scopes=["myscope"], ), )

from mcp.etdi.inspector import SecurityAnalyzer analyzer = SecurityAnalyzer() # Analyze tool security result = await analyzer.analyze_tool(tool_definition) print(f"Security Score: {result.security_score}") print(f"Vulnerabilities: {result.vulnerabilities}")

from mcp.etdi.inspector import TokenDebugger debugger = TokenDebugger() # Debug JWT tokens debug_info = await debugger.debug_token(jwt_token) print(f"Token valid: {debug_info.valid}") print(f"Claims: {debug_info.claims}") print(f"Issues: {debug_info.issues}")

from mcp.etdi.inspector import OAuthValidator validator = OAuthValidator() # Validate OAuth configuration result = await validator.validate_provider("auth0", oauth_config) print(f"Configuration valid: {result.configuration_valid}") print(f"Provider reachable: {result.is_reachable}")

# main.py import contextlib from fastapi import FastAPI from mcp.echo import echo from mcp.math import math # Create a combined lifespan to manage both session managers @contextlib.asynccontextmanager async def lifespan(app: FastAPI): async with contextlib.AsyncExitStack() as stack: await stack.enter_async_context(echo.mcp.session_manager.run()) await stack.enter_async_context(math.mcp.session_manager.run()) yield app = FastAPI(lifespan=lifespan) app.mount("/echo", echo.mcp.streamable_http_app()) app.mount("/math", math.mcp.streamable_http_app())

from mcp.server.fastmcp import FastMCP app = FastMCP("Secure Server") # Tool requiring cryptographic request signatures @app.tool(  etdi=True,  etdi_require_request_signing=True,  etdi_permissions=["banking:transfer"] ) def transfer_funds(amount: float, to_account: str) -> str: """High-security tool requiring signed requests""" return f"Transferred ${amount} to {to_account}" # Initialize request signing verification app.initialize_request_signing()

# Standard MCP request { "method": "tools/call", "params": { "name": "my_tool", "arguments": {"param": "value"} } } # ETDI signed request (backward compatible) { "method": "tools/call", "params": { "name": "my_tool", "arguments": {"param": "value"}, "etdi_signature": "base64-encoded-signature", "etdi_timestamp": "2024-01-01T12:00:00Z", "etdi_key_id": "client-key-id", "etdi_algorithm": "RS256" } }

from mcp.server.fastmcp import FastMCP mcp = FastMCP("Echo") @mcp.resource("echo://{message}") def echo_resource(message: str) -> str: """Echo a message as a resource""" return f"Resource echo: {message}" @mcp.tool() def echo_tool(message: str) -> str: """Echo a message as a tool""" return f"Tool echo: {message}" @mcp.prompt() def echo_prompt(message: str) -> str: """Create an echo prompt""" return f"Please process this message: {message}"

import sqlite3 from mcp.server.fastmcp import FastMCP mcp = FastMCP("SQLite Explorer") @mcp.resource("schema://main") def get_schema() -> str: """Provide the database schema as a resource""" conn = sqlite3.connect("database.db") schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall() return "\n".join(sql[0] for sql in schema if sql[0]) @mcp.tool() def query_data(sql: str) -> str: """Execute SQL queries safely""" conn = sqlite3.connect("database.db") try: result = conn.execute(sql).fetchall() return "\n".join(str(row) for row in result) except Exception as e: return f"Error: {str(e)}"

from contextlib import asynccontextmanager from collections.abc import AsyncIterator from fake_database import Database # Replace with your actual DB type from mcp.server import Server @asynccontextmanager async def server_lifespan(server: Server) -> AsyncIterator[dict]: """Manage server startup and shutdown lifecycle.""" # Initialize resources on startup db = await Database.connect() try: yield {"db": db} finally: # Clean up on shutdown await db.disconnect() # Pass lifespan to server server = Server("example-server", lifespan=server_lifespan) # Access lifespan context in handlers @server.call_tool() async def query_db(name: str, arguments: dict) -> list: ctx = server.request_context db = ctx.lifespan_context["db"] return await db.query(arguments["query"])

import mcp.server.stdio import mcp.types as types from mcp.server.lowlevel import NotificationOptions, Server from mcp.server.models import InitializationOptions # Create a server instance server = Server("example-server") @server.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: return [ types.Prompt( name="example-prompt", description="An example prompt template", arguments=[ types.PromptArgument( name="arg1", description="Example argument", required=True ) ], ) ] @server.get_prompt() async def handle_get_prompt( name: str, arguments: dict[str, str] | None ) -> types.GetPromptResult: if name != "example-prompt": raise ValueError(f"Unknown prompt: {name}") return types.GetPromptResult( description="Example prompt", messages=[ types.PromptMessage( role="user", content=types.TextContent(type="text", text="Example prompt text"), ) ], ) async def run(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="example", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if __name__ == "__main__": import asyncio asyncio.run(run())

from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client # Create server parameters for stdio connection server_params = StdioServerParameters( command="python", # Executable args=["example_server.py"], # Optional command line arguments env=None, # Optional environment variables ) # Optional: create a sampling callback async def handle_sampling_message( message: types.CreateMessageRequestParams, ) -> types.CreateMessageResult: return types.CreateMessageResult( role="assistant", content=types.TextContent( type="text", text="Hello, world! from model", ), model="gpt-3.5-turbo", stopReason="endTurn", ) async def run(): async with stdio_client(server_params) as (read, write): async with ClientSession( read, write, sampling_callback=handle_sampling_message ) as session: # Initialize the connection await session.initialize() # List available prompts prompts = await session.list_prompts() # Get a prompt prompt = await session.get_prompt( "example-prompt", arguments={"arg1": "value"} ) # List available resources resources = await session.list_resources() # List available tools tools = await session.list_tools() # Read a resource content, mime_type = await session.read_resource("file://some/path") # Call a tool result = await session.call_tool("tool-name", arguments={"arg1": "value"}) if __name__ == "__main__": import asyncio asyncio.run(run())

from mcp.client.streamable_http import streamablehttp_client from mcp import ClientSession async def main(): # Connect to a streamable HTTP server async with streamablehttp_client("example/mcp") as ( read_stream, write_stream, _, ): # Create a session using the client streams async with ClientSession(read_stream, write_stream) as session: # Initialize the connection await session.initialize() # Call a tool tool_result = await session.call_tool("echo", {"message": "hello"})

from mcp.client.auth import OAuthClientProvider, TokenStorage from mcp.client.session import ClientSession from mcp.client.streamable_http import streamablehttp_client from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken class CustomTokenStorage(TokenStorage): """Simple in-memory token storage implementation.""" async def get_tokens(self) -> OAuthToken | None: pass async def set_tokens(self, tokens: OAuthToken) -> None: pass async def get_client_info(self) -> OAuthClientInformationFull | None: pass async def set_client_info(self, client_info: OAuthClientInformationFull) -> None: pass async def main(): # Set up OAuth authentication oauth_auth = OAuthClientProvider( server_url="https://api.example.com", client_metadata=OAuthClientMetadata( client_name="My Client", redirect_uris=["http://localhost:3000/callback"], grant_types=["authorization_code", "refresh_token"], response_types=["code"], ), storage=CustomTokenStorage(), redirect_handler=lambda url: print(f"Visit: {url}"), callback_handler=lambda: ("auth_code", None), ) # Use with streamable HTTP client async with streamablehttp_client( "https://api.example.com/mcp", auth=oauth_auth ) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() # Authenticated session ready