LLM MCP 协议实战

def _fetch_papers(topic: str, max_results: int):
    client = arxiv.Client()
    search = arxiv.Search(query=topic, max_results=max_results,
                          sort_by=arxiv.SortCriterion.Relevance)
    return list(client.results(search))

@mcp.tool()
def search_papers(topic: str, max_results: int = 5) -> list[str]:
    path = _paper_info_path(topic)                       # papers/<topic>/papers_info.json
    papers_info = _load_papers_info(path)
    ids = []
    for paper in _fetch_papers(topic, max_results):
        short_id = paper.get_short_id()
        ids.append(short_id)
        papers_info[short_id] = extract_paper_info(paper) # 拍平 title/authors/summary/pdf_url/published
    _save_papers_info(path, papers_info)
    return ids

@mcp.tool()
def extract_info(paper_id: str) -> str:
    """Search for information about a specific paper across all topic directories."""
    for item in os.listdir(PAPER_DIR):
        item_path = os.path.join(PAPER_DIR, item)
        if not os.path.isdir(item_path): continue
        file_path = os.path.join(item_path, "papers_info.json")
        if not os.path.isfile(file_path): continue
        papers_info = _load_papers_info(file_path)
        if paper_id in papers_info:
            return json.dumps(papers_info[paper_id], indent=2)
    return f"There's no saved information related to paper {paper_id}."

@mcp.prompt()
def generate_search_prompt(topic: str, num_papers: int = 5) -> str:
    """Generate a prompt for Claude to find and discuss academic papers on a specific topic."""
    return f"""Search for {num_papers} academic papers about '{topic}' using the search_papers tool.
Follow these instructions:
1. First, search for papers using search_papers(topic='{topic}', max_results={num_papers})
2. For each paper found, extract and organize the following information:
   - Paper title / Authors / Publication date
   - Brief summary of the key findings
   - Main contributions or innovations
   - Methodologies used / Relevance to the topic '{topic}'
3. Provide a comprehensive summary ...

Use clear formatting with headers, bullet points for easy readability."""

import asyncio, sys
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

_MCP_SERVER = Path(__file__).parent / "papers.py"   # 绝对路径,与 CWD 无关

class MCP_ChatBot:
    def _openai_tools(self) -> list[dict]:
        """mcp 工具转 OpenAI function-calling 格式"""
        return [
            {"type": "function",
             "function": {"name": t["name"],
                          "description": t["description"],
                          "parameters": t["input_schema"]}}
            for t in self.available_tools
        ]

    async def connect_to_server_and_run(self):
        server_params = StdioServerParameters(
            command="uv",
            args=["run", str(_MCP_SERVER)],          # 绝对路径启动子进程
            env=None,
        )
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                self.session = session
                await session.initialize()
                response = await session.list_tools()
                self.available_tools = [
                    {"name": t.name, "description": t.description,
                     "input_schema": t.inputSchema}
                    for t in response.tools
                ]
                await self.chat_loop()

async def process_query(self, query):
    messages = [{"role": "user", "content": query}]
    llm_with_tools = self.llm.bind_tools(self._openai_tools())

    while True:
        response: AIMessage = await llm_with_tools.ainvoke(messages)
        messages.append(response)

        tool_calls = response.tool_calls or []
        if not tool_calls:
            if response.content: print(response.content)
            return                                     # 没有工具调用 → 直接回答

        for tc in tool_calls:
            result = await self.session.call_tool(tc["name"], arguments=tc["args"])
            content = "\n".join(
                getattr(block, "text", str(block)) for block in (result.content or [])
            )
            messages.append(ToolMessage(content=content, tool_call_id=tc["id"]))
        # 继续 while, 让 LLM 看到工具结果后再决定下一步

async def make_nws_request(url: str) -> dict[str, Any] | None:
    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
    async with httpx.AsyncClient() as client:
        try:
            r = await client.get(url, headers=headers, timeout=30.0)
            r.raise_for_status()
            return r.json()
        except Exception:
            return None

@mcp.tool()
async def get_alerts(state: str) -> str:
    """获取美国某个州当前生效的天气预警信息。state: 两个字母的美国州代码 (例如: CA, NY)。"""
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)
    if not data or "features" not in data:
        return "无法获取预警信息或未找到相关数据。"
    if not data["features"]:
        return "该州当前没有生效的天气预警。"
    alerts = [format_alert(feature) for feature in data["features"]]
    return "\n---\n".join(alerts)

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """根据经纬度获取天气预报。"""
    # 步骤 1: 经纬度 → forecast 端点 URL
    points = await make_nws_request(f"{NWS_API_BASE}/points/{latitude},{longitude}")
    if not points: return "无法获取该地点的预报数据。"

    # 步骤 2: 拉真正的预报数据
    forecast = await make_nws_request(points["properties"]["forecast"])
    if not forecast: return "无法获取详细的预报信息。"

    # 步骤 3: 取前 5 个 period 渲染成可读字符串
    periods = forecast["properties"]["periods"]
    return "\n---\n".join(
        f"{p['name']}: 温度 {p['temperature']}°{p['temperatureUnit']}"
        f" 风力 {p['windSpeed']} {p['windDirection']} {p['detailedForecast']}"
        for p in periods[:5]
    )

# mcp_lab/weather/weather.py
mcp = FastMCP("weather", host="0.0.0.0", port=8093)

if __name__ == "__main__":
    mcp.run(transport='sse')    # 远程部署用 sse 方式