File size: 3,618 Bytes
5da4770 | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | from fastapi import APIRouter, HTTPException, Depends, Query
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, validator, HttpUrl
from utils.auth_utils import get_current_user_id_from_jwt
from utils.logger import logger
from .infrastructure.dependencies import get_mcp_dependencies
from .domain.exceptions import MCPException, MCPServerNotFoundError, MCPToolNotFoundError
router = APIRouter()
def get_mcp_manager():
dependencies = get_mcp_dependencies()
return dependencies.mcp_manager
class MCPServerResponse(BaseModel):
qualified_name: str
display_name: str
description: str
created_at: str
use_count: int
homepage: str
icon_url: Optional[str] = None
is_deployed: Optional[bool] = None
tools: Optional[List[Dict[str, Any]]] = None
security: Optional[Dict[str, Any]] = None
class MCPServerListResponse(BaseModel):
servers: List[MCPServerResponse]
pagination: Dict[str, int]
class MCPServerDetailResponse(BaseModel):
qualified_name: str
display_name: str
icon_url: Optional[str] = None
deployment_url: Optional[str] = None
connections: List[Dict[str, Any]]
security: Optional[Dict[str, Any]] = None
tools: Optional[List[Dict[str, Any]]] = None
class PopularServersResponse(BaseModel):
success: bool
servers: List[Dict[str, Any]]
categorized: Dict[str, List[Dict[str, Any]]]
total: int
category_count: int
pagination: Dict[str, int]
class CustomMCPConnectionRequest(BaseModel):
url: str
config: Optional[Dict[str, Any]] = {}
class CustomMCPConnectionResponse(BaseModel):
success: bool
qualified_name: str
display_name: str
tools: List[Dict[str, Any]]
config: Dict[str, Any]
url: str
message: str
class CustomMCPDiscoverRequest(BaseModel):
type: str
config: Dict[str, Any]
@router.get("/mcp/servers/{qualified_name:path}", response_model=MCPServerDetailResponse)
async def get_mcp_server_details(
qualified_name: str,
user_id: str = Depends(get_current_user_id_from_jwt)
):
try:
manager = get_mcp_manager()
server_detail = await manager.get_server_details(qualified_name)
return MCPServerDetailResponse(
qualified_name=server_detail.qualified_name,
display_name=server_detail.display_name,
icon_url=server_detail.icon_url,
deployment_url=server_detail.deployment_url,
connections=server_detail.connections,
security=server_detail.security,
tools=server_detail.tools
)
except MCPServerNotFoundError:
raise HTTPException(status_code=404, detail="MCP server not found")
except MCPException as e:
logger.error(f"Error getting MCP server details: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@router.post("/mcp/discover-custom-tools")
async def discover_custom_mcp_tools(request: CustomMCPDiscoverRequest):
try:
manager = get_mcp_manager()
result = await manager.discover_custom_tools(request.type, request.config)
return CustomMCPConnectionResponse(
success=result.success,
qualified_name=result.qualified_name,
display_name=result.display_name,
tools=result.tools,
config=result.config,
url=result.url,
message=result.message
)
except MCPException as e:
logger.error(f"Error discovering custom MCP tools: {str(e)}")
raise HTTPException(status_code=500, detail=str(e)) |