从零开始的 LLM Agent编程生活–MCP 篇

export interface Resource {
  uri: string;
  name: string;
  description?: string;
  mimeType?: string;
  annotations?: Annotations;
  size?: number;
}

A detailed tool for dynamic and reflective problem-solving through thoughts.
This tool helps analyze problems through a flexible thinking process that can adapt and evolve.
Each thought can build on, question, or revise previous insights as understanding deepens.

When to use this tool:
- Breaking down complex problems into steps
- Planning and design with room for revision
- Analysis that might need course correction
- Problems where the full scope might not be clear initially
- Problems that require a multi-step solution
- Tasks that need to maintain context over multiple steps
- Situations where irrelevant information needs to be filtered out

Key features:
- You can adjust total_thoughts up or down as you progress
- You can question or revise previous thoughts
- You can add more thoughts even after reaching what seemed like the end
- You can express uncertainty and explore alternative approaches
- Not every thought needs to build linearly - you can branch or backtrack
- Generates a solution hypothesis
- Verifies the hypothesis based on the Chain of Thought steps
- Repeats the process until satisfied
- Provides a correct answer

Parameters explained:
- thought: Your current thinking step, which can include:
* Regular analytical steps
* Revisions of previous thoughts
* Questions about previous decisions
* Realizations about needing more analysis
* Changes in approach
* Hypothesis generation
* Hypothesis verification
- next_thought_needed: True if you need more thinking, even if at what seemed like the end
- thought_number: Current number in sequence (can go beyond initial total if needed)
- total_thoughts: Current estimate of thoughts needed (can be adjusted up/down)
- is_revision: A boolean indicating if this thought revises previous thinking
- revises_thought: If is_revision is true, which thought number is being reconsidered
- branch_from_thought: If branching, which thought number is the branching point
- branch_id: Identifier for the current branch (if any)
- needs_more_thoughts: If reaching end but realizing more thoughts needed

You should:
1. Start with an initial estimate of needed thoughts, but be ready to adjust
2. Feel free to question or revise previous thoughts
3. Don't hesitate to add more thoughts if needed, even at the "end"
4. Express uncertainty when present
5. Mark thoughts that revise previous thinking or branch into new paths
6. Ignore information that is irrelevant to the current step
7. Generate a solution hypothesis when appropriate
8. Verify the hypothesis based on the Chain of Thought steps
9. Repeat the process until satisfied with the solution
10. Provide a single, ideally correct answer as the final output
11. Only set next_thought_needed to false when truly done and a satisfactory answer is reached

import os
import queue
import threading
import logging
from typing import Any, Callable, Literal

import json

from urllib.parse import urljoin
from pydantic import BaseModel, ConfigDict
import requests

logger = logging.getLogger(__name__)


class Message(BaseModel):
    event_type: Literal["message"]
    data: str


class JsonRpcHeader(BaseModel):
    method: str
    jsonrpc: str = "2.0"


class JsonRpcRequest(JsonRpcHeader):
    params: dict | None = None
    id: int


class JsonRpcNotify(JsonRpcHeader): ...


type JsonRpcMessage = JsonRpcHeader | JsonRpcNotify


class JSONRPCResponse(BaseModel):
    """A successful (non-error) response to a request."""

    jsonrpc: str
    id: int
    result: dict[str, Any]
    model_config = ConfigDict(extra="allow")


class SSEClient(threading.Thread):
    def __init__(self, url: str) -> None:
        super().__init__()
        self._url = url
        self._queue = queue.Queue()
        self._ready_event = threading.Event()
        self._endpoint = ""
        self._callback = {}

    def run(self) -> None:
        print("run")
        self._event_loop()

    def wait_ready_for_endpoint(self) -> str:
        self._ready_event.wait()
        return self._endpoint

    def register_callback(
        self, message_type: str, callback: Callable[[Message], None]
    ) -> None:
        self._callback[message_type] = callback

    def _event_loop(self) -> None:
        response = requests.get(
            self._url,
            stream=True,
        )

        if response.status_code not in (200, 202):
            raise Exception(f"Failed to connect to server: {response.status_code}")

        event_type = None
        event_data = None

        for line in response.iter_lines(chunk_size=2, decode_unicode=True):
            print(line)
            if line.startswith("event:"):
                event_type = line[6:].strip()
            if line.startswith("data:"):
                event_data = line[5:].strip()

            if event_data is not None and event_type is not None:
                match event_type:
                    case "message":
                        msg = Message(event_type=event_type, data=event_data)
                        if "message" in self._callback:
                            try:
                                self._callback["message"](msg)
                            except Exception as e:
                                logger.error(e)
                        else:
                            self._queue.put(msg)

                    case "endpoint":
                        event_data = event_data.strip()
                        self._endpoint = event_data
                        self._ready_event.set()

                event_type, event_data = None, None


class McpClient:
    def __init__(
        self, sse: SSEClient, endpoint: str, session: requests.Session
    ) -> None:
        self._endpoint = endpoint
        self._sse = sse
        self._sess = session
        self.capabilities = {}
        self._id = 0
        self._callback_record = {}
        self._sse.register_callback("message", self.notify_callback)

    def initialize(self) -> None:
        result: dict = self._send_request(
            method="initialize",
            params={
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {"call": True}, "resources": {"read": True}},
                "clientInfo": dict(name="mcp", version="0.1.0"),
            },
        )
        self.capabilities = result.get("capabilities", {})
        self._send_request(method="notifications/initialized")

    def get_tools(self) -> dict:
        return self._send_request(method="tools/list")
    
    def call_tool(self, tool_name: str, params: dict) -> dict:
        return self._send_request(
            method="tools/call",
            params={
                "name": tool_name,
                "arguments": params,
            },
        )

    def notify_callback(self, message: Message) -> None:
        response = JSONRPCResponse.model_validate(json.loads(message.data))
        if response.id in self._callback_record:
            self._callback_record[response.id](response)

    def _send_request(self, method: str, params: dict | None = None) -> dict | None:

        event = threading.Event()
        result = None
        if "notifications" in method:
            data = JsonRpcNotify(
                method=method,
            )
        else:
            data = JsonRpcRequest(
                method=method,
                params=params,
                id=self._id,
            )

            def get_result(item):
                nonlocal result
                event.set()
                result = item.result

            self._callback_record[self._id] = get_result
            self._id += 1

        print(f"request: url:{self._endpoint} data:{data.json()}")

        response = self._sess.post(
            url=self._endpoint,
            json=data.model_dump(
                mode="json",
                exclude_none=True,
            ),
        )

        if response.status_code not in (200, 202):
            raise Exception(f"Connect Error: {response.text}")

        if "notifications" in method:
            return None

        event.wait()
        return result




def main() -> None:
    base = "https://mcp.amap.com/sse?key=" + os.environ["AMAP_KEY"]

    sse_client = SSEClient(url=base)
    sse_client.start()
    endpoint = sse_client.wait_ready_for_endpoint()

    mcp = McpClient(
        sse=sse_client, endpoint=urljoin(base, endpoint), session=requests.Session()
    )
    mcp.initialize()
    print("init Ok!")
    tools = mcp.get_tools()
    for tool in tools['tools']:
        print(f"tool: {tool['name']}")
        print(f"  description: {tool['description']}")
        print(f"  parameters: {json.dumps(tool['inputSchema'], indent=2, ensure_ascii=False)}")
    
    print("get tools Ok!")
    
    # Call a tool
    result = mcp.call_tool("maps_weather", {"city": "北京"})
    print("call_tool Ok!")
    print(f"result: {json.dumps(result, indent=2, ensure_ascii=False)}")
    
    exit()


if __name__ == "__main__":
    main()