# 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