FastMCP OAuth Server

""" MCP OAuth Client A client implementation for connecting to MCP servers with OAuth authentication. """ import asyncio import logging from typing import Dict, Any, Optional from urllib.parse import urlencode import httpx from mcp import ClientSession from mcp.client.session import ClientSession as MCPClientSession from client.auth_handler import OAuthHandler logger = logging.getLogger(__name__) class MCPOAuthClient: """ MCP client with OAuth authentication support. This client handles the OAuth flow for authenticating with MCP servers that require OAuth authentication. """ def __init__( self, server_url: str, client_id: Optional[str] = None, redirect_uri: str = "http://localhost:8080/callback", ): """ Initialize the MCP OAuth client. Args: server_url: URL of the MCP server client_id: OAuth client ID (if pre-registered) redirect_uri: OAuth redirect URI for authentication """ self.server_url = server_url.rstrip('/') self.client_id = client_id self.redirect_uri = redirect_uri self.auth_handler = OAuthHandler( server_url=server_url, client_id=client_id, redirect_uri=redirect_uri ) self.session: Optional[MCPClientSession] = None self.access_token: Optional[str] = None async def authenticate(self) -> bool: """ Perform OAuth authentication with the MCP server. Returns: True if authentication successful, False otherwise """ try: self.access_token = await self.auth_handler.authenticate() return self.access_token is not None except Exception as e: logger.error(f"Authentication failed: {e}") return False async def connect(self) -> bool: """ Connect to the MCP server using authenticated session. Returns: True if connection successful, False otherwise """ if not self.access_token: logger.error("No access token available. Please authenticate first.") return False try: # Create HTTP headers with bearer token headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json" } # TODO: Initialize MCP session with authenticated transport logger.info(f"Connecting to MCP server at {self.server_url}") # For now, we'll create a placeholder session # In a real implementation, you'd use the MCP client library # with the authenticated transport return True except Exception as e: logger.error(f"Failed to connect to MCP server: {e}") return False async def list_tools(self) -> Dict[str, Any]: """ List available tools from the MCP server. Returns: Dictionary containing available tools """ if not self.access_token: raise ValueError("Not authenticated. Call authenticate() first.") # TODO: Implement actual MCP tool listing # This would use the MCP protocol to list available tools logger.info("Listing available tools from MCP server") return { "tools": [ {"name": "get_weather", "description": "Get weather information"}, {"name": "get_user_info", "description": "Get user information"}, {"name": "protected_action", "description": "Perform protected action"} ] } async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """ Call a tool on the MCP server. Args: tool_name: Name of the tool to call arguments: Arguments to pass to the tool Returns: Tool execution result """ if not self.access_token: raise ValueError("Not authenticated. Call authenticate() first.") # TODO: Implement actual MCP tool calling # This would use the MCP protocol to call tools logger.info(f"Calling tool '{tool_name}' with arguments: {arguments}") # Placeholder response return { "tool": tool_name, "arguments": arguments, "result": f"Mock result for {tool_name}", "status": "success" } async def disconnect(self): """Disconnect from the MCP server.""" if self.session: # TODO: Properly close MCP session self.session = None self.access_token = None logger.info("Disconnected from MCP server") # Example usage async def main(): """Example usage of the MCP OAuth client.""" client = MCPOAuthClient("http://localhost:8000") print("🔐 Authenticating with MCP server...") if await client.authenticate(): print("✅ Authentication successful!") print("🔌 Connecting to MCP server...") if await client.connect(): print("✅ Connected successfully!") # List available tools tools = await client.list_tools() print(f"📋 Available tools: {tools}") # Call a tool result = await client.call_tool("get_weather", {"city": "London"}) print(f"🌤️ Weather result: {result}") await client.disconnect() else: print("❌ Authentication failed") if __name__ == "__main__": asyncio.run(main())