MCP 服务器

[features]
# ... other feature flags

[features.mcp.sse]
    # Enable or disable the SSE connection type globally
    enabled = true

[features.mcp.stdio]
    # Enable or disable the stdio connection type globally
    enabled = true
    # Define an allowlist of executables for the stdio type.
    # Only the base names of executables listed here can be used.
    # This is a crucial security measure for stdio connections.
    # Example: allows running `npx ...` and `uvx ...` but blocks others.
    allowed_executables = [ "npx", "uvx" ]

import chainlit as cl
from mcp import ClientSession

@cl.on_mcp_connect
async def on_mcp_connect(connection, session: ClientSession):
    """Called when an MCP connection is established"""
    # Your connection initialization code here
    # This handler is required for MCP to work
    
@cl.on_mcp_disconnect
async def on_mcp_disconnect(name: str, session: ClientSession):
    """Called when an MCP connection is terminated"""
    # Your cleanup code here
    # This handler is optional

@cl.on_mcp_connect
async def on_mcp(connection, session: ClientSession):
    # List available tools
    result = await session.list_tools()
    
    # Process tool metadata
    tools = [{
        "name": t.name,
        "description": t.description,
        "input_schema": t.inputSchema,
    } for t in result.tools]
    
    # Store tools for later use
    mcp_tools = cl.user_session.get("mcp_tools", {})
    mcp_tools[connection.name] = tools
    cl.user_session.set("mcp_tools", mcp_tools)

@cl.step(type="tool") 
async def call_tool(tool_use):
    tool_name = tool_use.name
    tool_input = tool_use.input
    
    # Find appropriate MCP connection for this tool
    mcp_name = find_mcp_for_tool(tool_name)
    
    # Get the MCP session
    mcp_session, _ = cl.context.session.mcp_sessions.get(mcp_name)
    
    # Call the tool
    result = await mcp_session.call_tool(tool_name, tool_input)
    
    return result

async def call_model_with_tools():
    # Get tools from all MCP connections
    mcp_tools = cl.user_session.get("mcp_tools", {})
    all_tools = [tool for connection_tools in mcp_tools.values() for tool in connection_tools]
    
    # Call your LLM with the tools
    response = await your_llm_client.call(
        messages=messages,
        tools=all_tools
    )
    
    # Handle tool calls if needed
    if response.has_tool_calls():
        # Process tool calls
        pass
        
    return response