File size: 2,201 Bytes
8791d59 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
from typing import Any, List, Optional
import json
from fastapi import FastAPI, HTTPException
from mcp import Server, Resource, Tool
from mcp.types import TextContent, ImageContent
import logging
from .resources import get_resources
logger = logging.getLogger(__name__)
class MCPServer:
def __init__(self, app: FastAPI):
self.app = app
self.server = Server()
self._setup_resources()
self._setup_tools()
def _setup_resources(self):
"""Définir les ressources exposées via MCP"""
resources = get_resources()
for resource in resources:
self.server.add_resource(resource)
def _setup_tools(self):
"""Définir les outils exposés via MCP"""
from .tools import get_tools
tools = get_tools()
for tool in tools:
self.server.add_tool(tool)
async def list_resources(self) -> List[dict]:
"""Lister toutes les ressources disponibles"""
return [
{
"uri": resource.uri,
"name": resource.name,
"description": resource.description,
"mime_type": resource.mime_type
}
for resource in self.server.resources.values()
]
async def list_tools(self) -> List[dict]:
"""Lister tous les outils disponibles"""
return [
{
"name": tool.name,
"description": tool.description,
"input_schema": tool.input_schema
}
for tool in self.server.tools.values()
]
async def call_tool(self, tool_name: str, arguments: dict) -> Any:
"""Appeler un outil MCP"""
if tool_name not in self.server.tools:
raise HTTPException(status_code=404, detail=f"Tool {tool_name} not found")
try:
tool = self.server.tools[tool_name]
result = await tool.execute(arguments)
return result
except Exception as e:
logger.error(f"Tool execution error: {str(e)}")
raise HTTPException(status_code=500, detail=f"Tool execution failed: {str(e)}") |