MCP 快速开始

# server.py
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

# To run as an SSE server, you could initialize like this:
mcp = FastMCP(name="demo", host="127.0.0.1", port=8256, sse_path="/sse")
# mcp = FastMCP()

@mcp.tool()
def add_2_numbers(a: int, b: int) -> int:
    """Adds two numbers together."""
    return a + b

@mcp.resource("config://app")
def get_config() -> str:
    """Static configuration data."""
    return "App configuration here"

@mcp.prompt()
def debug_error(error: str) -> list[base.Message]:
    """A sample prompt to help debug an error."""
    return [
        base.UserMessage("I'm seeing this error:"),
        base.UserMessage(error),
        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
    ]

@mcp.tool()
def multiply_2_numbers(a: int, b: int) -> int:
    """Multiplies two numbers."""
    return a * b

if __name__ == "__main__":
    # You can choose the transport method here.
    # To run as an SSE server, use: mcp.run(transport='sse')
    # To run as a stdio server, use: mcp.run(transport='stdio')
    print("Starting MCP server with stdio transport...")
    mcp.run(transport='sse')

# client.py
import asyncio
import json
import os

os.chdir(os.path.dirname(os.path.abspath(__file__)))

from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, StdioServerParameters, stdio_client
from mcp.client.sse import sse_client
from openai import AsyncOpenAI

class GenericMCPClient:
    """
    A generic MCP client that can connect to an MCP server
    via stdio or sse, and interact with an LLM to use the server's tools.
    """
    def __init__(self, api_key: str, base_url: str, model: str):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)
        self.model = model
        self.messages = []
        self.available_tools = []

    async def connect_to_stdio_server(self, command: str, args: list, env: dict = None):
        """Connects to a local MCP server using stdio."""
        print("Connecting to stdio server...")
        server_params = StdioServerParameters(command=command, args=args, env=env or {})
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        read, write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(read, write))
        await self._initialize_session("stdio")

    async def connect_to_sse_server(self, server_url: str, headers: dict = None):
        """Connects to a remote MCP server using sse."""
        print(f"Connecting to SSE server at {server_url}...")
        sse_transport = await self.exit_stack.enter_async_context(sse_client(server_url, headers))
        read, write = sse_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(read, write))
        await self._initialize_session("sse")

    async def _initialize_session(self, connection_type: str):
        """Initializes the client session and lists available tools."""
        await self.session.initialize()
        response = await self.session.list_tools()
        tools = response.tools
        print(f"Successfully connected via {connection_type}. Available tools: {[tool.name for tool in tools]}")
        self.available_tools = [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                },
            }
            for tool in tools
        ]

    async def process_query(self, query: str):
        """Processes a user query by potentially calling tools via an LLM."""
        self.messages.append({"role": "user", "content": query})

        while True:
            response = await self.client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.available_tools
            )

            assistant_message = response.choices[0].message
            self.messages.append(assistant_message)

            if not assistant_message.tool_calls:
                return assistant_message.content

            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                print(f"Calling tool '{tool_name}' with args: {tool_args}")
                result = await self.session.call_tool(tool_name, tool_args)
                tool_output = result.content[0].text
                print(f"Tool '{tool_name}' result: {tool_output}")

                self.messages.append({
                    "role": "tool",
                    "content": tool_output,
                    "tool_call_id": tool_call.id
                })

    async def chat_loop(self):
        """Main chat loop to interact with the user."""
        self.messages = []
        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == 'quit':
                    break
                if not query:
                    continue
                response_text = await self.process_query(query)
                print("\nAI:", response_text)
            except Exception as e:
                print(f"\nAn error occurred: {e}")

    async def cleanup(self):
        """Cleans up resources by closing the async exit stack."""
        print("Cleaning up and closing connections...")
        await self.exit_stack.aclose()

async def main():
    # Load configuration from a file (e.g., config.json)
    # The config file should contain your LLM API key, base URL, and model name.
    try:
        with open("config.json", "r") as f:
            config = json.load(f)
    except FileNotFoundError:
        print("Error: config.json not found. Please create it with your LLM configuration.")
        return

    client = GenericMCPClient(config["llm"]["api_key"], config["llm"]["base_url"], config["llm"]["model"])

    try:
        # --- Choose your connection method ---
        # 1. Connect via stdio
        # await client.connect_to_stdio_server("python", ["server.py"])

        # 2. Connect via sse (uncomment to use)
        # Make sure your server.py is running in SSE mode on the correct host/port.
        await client.connect_to_sse_server("http://127.0.0.1:8256/sse")

        await client.chat_loop()
    except Exception as e:
        print(f"A critical error occurred: {e}")
    finally:
        await client.cleanup()

if __name__ == '__main__':
    asyncio.run(main())