Foxxy Bridge

# # MCP Foxxy Bridge - Proxy Server # # Copyright (C) 2024 Billy Bryant # Portions copyright (C) 2024 Sergey Parfenyuk (original MIT-licensed author) # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <https://www.gnu.org/licenses/>. # # MIT License attribution: Portions of this file were originally licensed # under the MIT License by Sergey Parfenyuk (2024). # """Create an MCP server that proxies requests through an MCP client. This server is created independent of any transport mechanism. """ from mcp import server, types from mcp.client.session import ClientSession from mcp_foxxy_bridge.utils.logging import get_logger logger = get_logger(__name__, facility="SERVER") async def create_proxy_server(remote_app: ClientSession) -> server.Server[object]: """Create a server instance from a remote app.""" logger.debug("Sending initialization request to remote MCP server...") response = await remote_app.initialize() capabilities = response.capabilities logger.debug("Configuring proxied MCP server...") app: server.Server[object] = server.Server(name=response.serverInfo.name) if capabilities.prompts: logger.debug("Capabilities: adding Prompts...") async def _list_prompts(_: types.ListPromptsRequest) -> types.ServerResult: result = await remote_app.list_prompts() return types.ServerResult(result) app.request_handlers[types.ListPromptsRequest] = _list_prompts async def _get_prompt(req: types.GetPromptRequest) -> types.ServerResult: result = await remote_app.get_prompt(req.params.name, req.params.arguments) return types.ServerResult(result) app.request_handlers[types.GetPromptRequest] = _get_prompt if capabilities.resources: logger.debug("Capabilities: adding Resources...") async def _list_resources(_: types.ListResourcesRequest) -> types.ServerResult: result = await remote_app.list_resources() return types.ServerResult(result) app.request_handlers[types.ListResourcesRequest] = _list_resources async def _list_resource_templates( _: types.ListResourceTemplatesRequest, ) -> types.ServerResult: result = await remote_app.list_resource_templates() return types.ServerResult(result) app.request_handlers[types.ListResourceTemplatesRequest] = _list_resource_templates async def _read_resource(req: types.ReadResourceRequest) -> types.ServerResult: result = await remote_app.read_resource(req.params.uri) return types.ServerResult(result) app.request_handlers[types.ReadResourceRequest] = _read_resource if capabilities.logging: logger.debug("Capabilities: adding Logging...") async def _set_logging_level(req: types.SetLevelRequest) -> types.ServerResult: await remote_app.set_logging_level(req.params.level) return types.ServerResult(types.EmptyResult()) app.request_handlers[types.SetLevelRequest] = _set_logging_level if capabilities.resources: logger.debug("Capabilities: adding Resources...") async def _subscribe_resource(req: types.SubscribeRequest) -> types.ServerResult: await remote_app.subscribe_resource(req.params.uri) return types.ServerResult(types.EmptyResult()) app.request_handlers[types.SubscribeRequest] = _subscribe_resource async def _unsubscribe_resource(req: types.UnsubscribeRequest) -> types.ServerResult: await remote_app.unsubscribe_resource(req.params.uri) return types.ServerResult(types.EmptyResult()) app.request_handlers[types.UnsubscribeRequest] = _unsubscribe_resource if capabilities.tools: logger.debug("Capabilities: adding Tools...") async def _list_tools(_: types.ListToolsRequest) -> types.ServerResult: tools = await remote_app.list_tools() return types.ServerResult(tools) app.request_handlers[types.ListToolsRequest] = _list_tools async def _call_tool(req: types.CallToolRequest) -> types.ServerResult: try: result = await remote_app.call_tool( req.params.name, (req.params.arguments or {}), ) return types.ServerResult(result) except Exception as e: return types.ServerResult( types.CallToolResult( content=[types.TextContent(type="text", text=str(e))], isError=True, ), ) app.request_handlers[types.CallToolRequest] = _call_tool async def _send_progress_notification(req: types.ProgressNotification) -> None: await remote_app.send_progress_notification( req.params.progressToken, req.params.progress, req.params.total, ) app.notification_handlers[types.ProgressNotification] = _send_progress_notification async def _complete(req: types.CompleteRequest) -> types.ServerResult: result = await remote_app.complete( req.params.ref, req.params.argument.model_dump(), ) return types.ServerResult(result) app.request_handlers[types.CompleteRequest] = _complete return app