【案例】Qwen 微调实战 · MCP 协议与工具接入

"""一个最小可用的 MCP 服务器：查城市实时天气。

三个函数分工明确：
    fetch_weather(city)   —— 发 HTTP 请求，拿原始 JSON
    format_weather(data)  —— 把 JSON 揉成人读得懂的一段话
    query_weather(city)   —— 用 @mcp.tool() 注册出去，串起上面两步

API key 走环境变量，源码里一个字符都不留：
    export OPENWEATHER_API_KEY="你的 key"

依赖：pip install "mcp>=1.6.0" httpx python-dotenv
运行：python weather_server.py      # 以 stdio 方式等待客户端连接
"""

import json
import os
from typing import Any

import httpx
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# 服务器名字会随工具列表一起暴露给客户端
mcp = FastMCP("WeatherServer")

load_dotenv()

OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT = "weather-app/1.0"


def _api_key() -> str:
    """每次调用时现取 key，取不到就直接报错，绝不带着空 key 去发请求。"""
    key = os.environ.get("OPENWEATHER_API_KEY")
    if not key:
        raise RuntimeError("未找到 OPENWEATHER_API_KEY，请在 .env 或环境变量中配置")
    return key


async def fetch_weather(city: str) -> dict[str, Any]:
    """异步请求 OpenWeather，返回原始字典；出错时返回带 error 键的字典。"""
    params = {
        "q": city,             # 城市英文名，例如 Beijing
        "appid": _api_key(),
        "units": "metric",     # 公制，温度用摄氏度
        "lang": "zh_cn",       # 天气描述返回中文
    }
    headers = {"User-Agent": USER_AGENT}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                OPENWEATHER_API_BASE,
                params=params,
                headers=headers,
                timeout=30.0,
            )
            # 非 2xx 直接抛出，交给下面的 except 收口
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP 错误: {e.response.status_code}"}
        except Exception as e:  # 网络超时、DNS 失败等
            return {"error": f"请求失败: {e}"}


def format_weather(data: dict[str, Any] | str) -> str:
    """把原始数据格式化成一段可读文本；字段缺失一律给默认值，不抛 KeyError。"""
    if isinstance(data, str):
        try:
            data = json.loads(data)
        except Exception as e:
            return f"无法解析天气数据: {e}"

    # 上游返回的错误原样转述，让模型知道这次查询失败了
    if "error" in data:
        return f"查询失败：{data['error']}"

    # 全部用 .get() 逐层取，任何一层缺失都退化成默认值
    city = data.get("name", "未知")
    country = data.get("sys", {}).get("country", "未知")
    temp = data.get("main", {}).get("temp", "N/A")
    humidity = data.get("main", {}).get("humidity", "N/A")
    wind_speed = data.get("wind", {}).get("speed", "N/A")
    weather_list = data.get("weather", [{}])
    description = weather_list[0].get("description", "未知")

    return (
        f"{city}, {country}\n"
        f"温度: {temp}°C\n"
        f"湿度: {humidity}%\n"
        f"风速: {wind_speed} m/s\n"
        f"天气: {description}\n"
    )


@mcp.tool()
async def query_weather(city: str) -> str:
    """查询指定城市的实时天气。

    :param city: 城市英文名称，例如 Beijing、Shanghai
    :return: 格式化后的天气信息
    """
    # 这段 docstring 会被 MCP 转成工具描述交给模型，写清楚用途和参数
    data = await fetch_weather(city)
    return format_weather(data)


if __name__ == "__main__":
    # stdio：客户端把本进程作为子进程拉起，用标准输入输出通信
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "TODO-允许访问的目录"]
    },
    "git": {
      "command": "uvx",
      "args": ["mcp-server-git", "--repository", "TODO-本地仓库路径"]
    }
  }
}

"""MCP 客户端：连多个 MCP 服务器，把它们的工具交给大模型，用 Function Call 调起来。

四个类各管一段：
    Configuration —— 读 .env 与 servers_config.json
    Server        —— 单个 MCP 服务器的全生命周期
    Tool          —— 把 MCP 工具描述转成模型读得懂的格式
    MultiServerMCPClient —— 汇总所有工具、跑对话循环

凭据全部来自环境变量，源码里不出现任何 key：
    LLM_API_KEY / BASE_URL / MODEL

依赖：pip install "mcp>=1.6.0" "openai>=1.76.2" python-dotenv
运行：python mcp_client.py
"""

import asyncio
import json
import logging
import os
from contextlib import AsyncExitStack
from typing import Any, Dict, List, Optional

from dotenv import load_dotenv
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)


class Configuration:
    """管理环境变量与服务器配置文件。"""

    def __init__(self) -> None:
        load_dotenv()
        self.api_key = os.environ.get("LLM_API_KEY")
        self.base_url = os.environ.get("BASE_URL")
        self.model = os.environ.get("MODEL")
        # 缺 key 直接拦在启动阶段，别等到第一次请求才 401
        if not self.api_key:
            raise ValueError("未找到 LLM_API_KEY，请在 .env 文件中配置")

    @staticmethod
    def load_config(file_path: str) -> Dict[str, Any]:
        with open(file_path, "r", encoding="utf-8") as f:
            return json.load(f)


class Tool:
    """封装 MCP 返回的工具信息，并生成给模型看的描述。"""

    def __init__(self, name: str, description: str,
                 input_schema: Dict[str, Any]) -> None:
        self.name = name
        self.description = description
        self.input_schema = input_schema

    def format_for_llm(self) -> str:
        args_desc = []
        for param_name, param_info in self.input_schema.get("properties", {}).items():
            line = f"- {param_name}: {param_info.get('description', 'No description')}"
            if param_name in self.input_schema.get("required", []):
                line += " (required)"
            args_desc.append(line)
        joined = "\n".join(args_desc)
        return f"Tool: {self.name}\nDescription: {self.description}\nArguments:\n{joined}"


class Server:
    """管理单个 MCP 服务器连接与工具调用。"""

    def __init__(self, name: str, config: Dict[str, Any]) -> None:
        self.name = name
        self.config = config
        self.session: Optional[ClientSession] = None
        # 所有异步资源压进同一个栈，退出时按相反顺序释放
        self.exit_stack = AsyncExitStack()
        # 防止并发清理同一批资源
        self._cleanup_lock = asyncio.Lock()

    async def initialize(self) -> None:
        command = self.config["command"]
        if not command:
            raise ValueError("command 不能为空")

        server_params = StdioServerParameters(
            command=command,
            args=self.config["args"],
            # 自定义 env 叠在系统环境变量之上，子进程才读得到 API key
            env={**os.environ, **self.config["env"]} if self.config.get("env") else None,
        )
        try:
            read_stream, write_stream = await self.exit_stack.enter_async_context(
                stdio_client(server_params))
            session = await self.exit_stack.enter_async_context(
                ClientSession(read_stream, write_stream))
            # 握手：交换协议版本与能力
            await session.initialize()
            self.session = session
        except Exception as e:
            logging.error("初始化服务器 %s 失败: %s", self.name, e)
            await self.cleanup()
            raise

    async def list_tools(self) -> List[Tool]:
        if not self.session:
            raise RuntimeError(f"服务器 {self.name} 尚未初始化")
        tools_response = await self.session.list_tools()
        tools: List[Tool] = []
        # 返回的是若干 (字段名, 值) 元组，只取 tools 那一条
        for item in tools_response:
            if isinstance(item, tuple) and item[0] == "tools":
                for tool in item[1]:
                    tools.append(Tool(tool.name, tool.description, tool.inputSchema))
        return tools

    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any],
                           retries: int = 2, delay: float = 1.0) -> Any:
        """执行工具，失败自动重试，重试耗尽才抛出。"""
        if not self.session:
            raise RuntimeError(f"服务器 {self.name} 尚未初始化")
        attempt = 0
        while attempt < retries:
            try:
                logging.info("在 %s 上执行 %s ...", self.name, tool_name)
                return await self.session.call_tool(tool_name, arguments)
            except Exception as e:
                attempt += 1
                logging.warning("执行工具出错: %s（第 %d/%d 次）", e, attempt, retries)
                if attempt < retries:
                    await asyncio.sleep(delay)
                else:
                    logging.error("重试次数用尽")
                    raise

    async def cleanup(self) -> None:
        async with self._cleanup_lock:
            try:
                await self.exit_stack.aclose()
                self.session = None
            except Exception as e:
                logging.error("清理服务器 %s 时出错: %s", self.name, e)


class MultiServerMCPClient:
    """连接多个 MCP 服务器，汇总工具，跑对话循环。"""

    def __init__(self) -> None:
        config = Configuration()
        self.client = OpenAI(api_key=config.api_key, base_url=config.base_url)
        self.model = config.model
        self.servers: Dict[str, Server] = {}
        self.all_tools: List[Dict[str, Any]] = []

    async def connect_to_servers(self, servers_config: Dict[str, Any]) -> None:
        for server_name, srv_config in servers_config.get("mcpServers", {}).items():
            server = Server(server_name, srv_config)
            await server.initialize()
            self.servers[server_name] = server
            for tool in await server.list_tools():
                # 统一重命名成 服务名_工具名，避免不同服务器的同名工具打架
                self.all_tools.append({
                    "type": "function",
                    "function": {
                        "name": f"{server_name}_{tool.name}",
                        "description": tool.description,
                        "parameters": {
                            "type": tool.input_schema.get("type", "object"),
                            "properties": tool.input_schema.get("properties", {}),
                            "required": tool.input_schema.get("required", []),
                        },
                    },
                })
        logging.info("已连接服务器: %s", ", ".join(self.servers) or "无")
        logging.info("可用工具: %s",
                     ", ".join(t["function"]["name"] for t in self.all_tools) or "无")

    async def _call_mcp_tool(self, full_name: str, args: Dict[str, Any]) -> str:
        server_name, _, tool_name = full_name.partition("_")
        server = self.servers.get(server_name)
        if not server:
            return f"找不到服务器: {server_name}"
        resp = await server.execute_tool(tool_name, args)
        return str(resp.content) if resp.content else "工具执行无输出"

    async def chat_base(self, messages: List[Dict[str, Any]]) -> Any:
        """一直循环到模型不再要求调用工具为止。"""
        response = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=self.all_tools)
        while response.choices[0].finish_reason == "tool_calls":
            # 把这轮 assistant 消息原样回填，再逐个补上工具回执
            messages.append(response.choices[0].message.model_dump())
            for call in response.choices[0].message.tool_calls:
                result = await self._call_mcp_tool(
                    call.function.name, json.loads(call.function.arguments))
                messages.append({
                    "role": "tool",
                    "content": result,
                    "tool_call_id": call.id,
                })
            response = self.client.chat.completions.create(
                model=self.model, messages=messages, tools=self.all_tools)
        return response

    async def chat_loop(self) -> None:
        print("MCP 客户端已启动，输入 quit 退出。")
        messages: List[Dict[str, Any]] = []
        while True:
            query = input("\n你: ").strip()
            if query.lower() == "quit":
                break
            try:
                messages.append({"role": "user", "content": query})
                messages = messages[-20:]        # 只留最近 20 条，防止上下文撑爆
                response = await self.chat_base(messages)
                messages.append(response.choices[0].message.model_dump())
                print(f"\nAI: {response.choices[0].message.content}")
            except Exception as e:
                print(f"\n调用过程出错: {e}")

    async def cleanup(self) -> None:
        for server in self.servers.values():
            await server.cleanup()


async def main() -> None:
    config = Configuration()
    servers_config = config.load_config("servers_config.json")
    client = MultiServerMCPClient()
    try:
        await client.connect_to_servers(servers_config)
        await client.chat_loop()
    finally:
        # 给子进程一点时间收尾，再统一释放
        await asyncio.sleep(0.1)
        await client.cleanup()


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

"""一个最小可用的 MCP 服务器：查城市实时天气。

三个函数分工明确：
    fetch_weather(city)   —— 发 HTTP 请求，拿原始 JSON
    format_weather(data)  —— 把 JSON 揉成人读得懂的一段话
    query_weather(city)   —— 用 @mcp.tool() 注册出去，串起上面两步

API key 走环境变量，源码里一个字符都不留：
    export OPENWEATHER_API_KEY="你的 key"

依赖：pip install "mcp>=1.6.0" httpx python-dotenv
运行：python weather_server.py      # 以 stdio 方式等待客户端连接
"""

import json
import os
from typing import Any

import httpx
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# 服务器名字会随工具列表一起暴露给客户端
mcp = FastMCP("WeatherServer")

load_dotenv()

OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT = "weather-app/1.0"


def _api_key() -> str:
    """每次调用时现取 key，取不到就直接报错，绝不带着空 key 去发请求。"""
    key = os.environ.get("OPENWEATHER_API_KEY")
    if not key:
        raise RuntimeError("未找到 OPENWEATHER_API_KEY，请在 .env 或环境变量中配置")
    return key


async def fetch_weather(city: str) -> dict[str, Any]:
    """异步请求 OpenWeather，返回原始字典；出错时返回带 error 键的字典。"""
    params = {
        "q": city,             # 城市英文名，例如 Beijing
        "appid": _api_key(),
        "units": "metric",     # 公制，温度用摄氏度
        "lang": "zh_cn",       # 天气描述返回中文
    }
    headers = {"User-Agent": USER_AGENT}

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                OPENWEATHER_API_BASE,
                params=params,
                headers=headers,
                timeout=30.0,
            )
            # 非 2xx 直接抛出，交给下面的 except 收口
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as e:
            return {"error": f"HTTP 错误: {e.response.status_code}"}
        except Exception as e:  # 网络超时、DNS 失败等
            return {"error": f"请求失败: {e}"}


def format_weather(data: dict[str, Any] | str) -> str:
    """把原始数据格式化成一段可读文本；字段缺失一律给默认值，不抛 KeyError。"""
    if isinstance(data, str):
        try:
            data = json.loads(data)
        except Exception as e:
            return f"无法解析天气数据: {e}"

    # 上游返回的错误原样转述，让模型知道这次查询失败了
    if "error" in data:
        return f"查询失败：{data['error']}"

    # 全部用 .get() 逐层取，任何一层缺失都退化成默认值
    city = data.get("name", "未知")
    country = data.get("sys", {}).get("country", "未知")
    temp = data.get("main", {}).get("temp", "N/A")
    humidity = data.get("main", {}).get("humidity", "N/A")
    wind_speed = data.get("wind", {}).get("speed", "N/A")
    weather_list = data.get("weather", [{}])
    description = weather_list[0].get("description", "未知")

    return (
        f"{city}, {country}\n"
        f"温度: {temp}°C\n"
        f"湿度: {humidity}%\n"
        f"风速: {wind_speed} m/s\n"
        f"天气: {description}\n"
    )


@mcp.tool()
async def query_weather(city: str) -> str:
    """查询指定城市的实时天气。

    :param city: 城市英文名称，例如 Beijing、Shanghai
    :return: 格式化后的天气信息
    """
    # 这段 docstring 会被 MCP 转成工具描述交给模型，写清楚用途和参数
    data = await fetch_weather(city)
    return format_weather(data)


if __name__ == "__main__":
    # stdio：客户端把本进程作为子进程拉起，用标准输入输出通信
    mcp.run(transport="stdio")

"""MCP 客户端：连多个 MCP 服务器，把它们的工具交给大模型，用 Function Call 调起来。

四个类各管一段：
    Configuration —— 读 .env 与 servers_config.json
    Server        —— 单个 MCP 服务器的全生命周期
    Tool          —— 把 MCP 工具描述转成模型读得懂的格式
    MultiServerMCPClient —— 汇总所有工具、跑对话循环

凭据全部来自环境变量，源码里不出现任何 key：
    LLM_API_KEY / BASE_URL / MODEL

依赖：pip install "mcp>=1.6.0" "openai>=1.76.2" python-dotenv
运行：python mcp_client.py
"""

import asyncio
import json
import logging
import os
from contextlib import AsyncExitStack
from typing import Any, Dict, List, Optional

from dotenv import load_dotenv
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)


class Configuration:
    """管理环境变量与服务器配置文件。"""

    def __init__(self) -> None:
        load_dotenv()
        self.api_key = os.environ.get("LLM_API_KEY")
        self.base_url = os.environ.get("BASE_URL")
        self.model = os.environ.get("MODEL")
        # 缺 key 直接拦在启动阶段，别等到第一次请求才 401
        if not self.api_key:
            raise ValueError("未找到 LLM_API_KEY，请在 .env 文件中配置")

    @staticmethod
    def load_config(file_path: str) -> Dict[str, Any]:
        with open(file_path, "r", encoding="utf-8") as f:
            return json.load(f)


class Tool:
    """封装 MCP 返回的工具信息，并生成给模型看的描述。"""

    def __init__(self, name: str, description: str,
                 input_schema: Dict[str, Any]) -> None:
        self.name = name
        self.description = description
        self.input_schema = input_schema

    def format_for_llm(self) -> str:
        args_desc = []
        for param_name, param_info in self.input_schema.get("properties", {}).items():
            line = f"- {param_name}: {param_info.get('description', 'No description')}"
            if param_name in self.input_schema.get("required", []):
                line += " (required)"
            args_desc.append(line)
        joined = "\n".join(args_desc)
        return f"Tool: {self.name}\nDescription: {self.description}\nArguments:\n{joined}"


class Server:
    """管理单个 MCP 服务器连接与工具调用。"""

    def __init__(self, name: str, config: Dict[str, Any]) -> None:
        self.name = name
        self.config = config
        self.session: Optional[ClientSession] = None
        # 所有异步资源压进同一个栈，退出时按相反顺序释放
        self.exit_stack = AsyncExitStack()
        # 防止并发清理同一批资源
        self._cleanup_lock = asyncio.Lock()

    async def initialize(self) -> None:
        command = self.config["command"]
        if not command:
            raise ValueError("command 不能为空")

        server_params = StdioServerParameters(
            command=command,
            args=self.config["args"],
            # 自定义 env 叠在系统环境变量之上，子进程才读得到 API key
            env={**os.environ, **self.config["env"]} if self.config.get("env") else None,
        )
        try:
            read_stream, write_stream = await self.exit_stack.enter_async_context(
                stdio_client(server_params))
            session = await self.exit_stack.enter_async_context(
                ClientSession(read_stream, write_stream))
            # 握手：交换协议版本与能力
            await session.initialize()
            self.session = session
        except Exception as e:
            logging.error("初始化服务器 %s 失败: %s", self.name, e)
            await self.cleanup()
            raise

    async def list_tools(self) -> List[Tool]:
        if not self.session:
            raise RuntimeError(f"服务器 {self.name} 尚未初始化")
        tools_response = await self.session.list_tools()
        tools: List[Tool] = []
        # 返回的是若干 (字段名, 值) 元组，只取 tools 那一条
        for item in tools_response:
            if isinstance(item, tuple) and item[0] == "tools":
                for tool in item[1]:
                    tools.append(Tool(tool.name, tool.description, tool.inputSchema))
        return tools

    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any],
                           retries: int = 2, delay: float = 1.0) -> Any:
        """执行工具，失败自动重试，重试耗尽才抛出。"""
        if not self.session:
            raise RuntimeError(f"服务器 {self.name} 尚未初始化")
        attempt = 0
        while attempt < retries:
            try:
                logging.info("在 %s 上执行 %s ...", self.name, tool_name)
                return await self.session.call_tool(tool_name, arguments)
            except Exception as e:
                attempt += 1
                logging.warning("执行工具出错: %s（第 %d/%d 次）", e, attempt, retries)
                if attempt < retries:
                    await asyncio.sleep(delay)
                else:
                    logging.error("重试次数用尽")
                    raise

    async def cleanup(self) -> None:
        async with self._cleanup_lock:
            try:
                await self.exit_stack.aclose()
                self.session = None
            except Exception as e:
                logging.error("清理服务器 %s 时出错: %s", self.name, e)


class MultiServerMCPClient:
    """连接多个 MCP 服务器，汇总工具，跑对话循环。"""

    def __init__(self) -> None:
        config = Configuration()
        self.client = OpenAI(api_key=config.api_key, base_url=config.base_url)
        self.model = config.model
        self.servers: Dict[str, Server] = {}
        self.all_tools: List[Dict[str, Any]] = []

    async def connect_to_servers(self, servers_config: Dict[str, Any]) -> None:
        for server_name, srv_config in servers_config.get("mcpServers", {}).items():
            server = Server(server_name, srv_config)
            await server.initialize()
            self.servers[server_name] = server
            for tool in await server.list_tools():
                # 统一重命名成 服务名_工具名，避免不同服务器的同名工具打架
                self.all_tools.append({
                    "type": "function",
                    "function": {
                        "name": f"{server_name}_{tool.name}",
                        "description": tool.description,
                        "parameters": {
                            "type": tool.input_schema.get("type", "object"),
                            "properties": tool.input_schema.get("properties", {}),
                            "required": tool.input_schema.get("required", []),
                        },
                    },
                })
        logging.info("已连接服务器: %s", ", ".join(self.servers) or "无")
        logging.info("可用工具: %s",
                     ", ".join(t["function"]["name"] for t in self.all_tools) or "无")

    async def _call_mcp_tool(self, full_name: str, args: Dict[str, Any]) -> str:
        server_name, _, tool_name = full_name.partition("_")
        server = self.servers.get(server_name)
        if not server:
            return f"找不到服务器: {server_name}"
        resp = await server.execute_tool(tool_name, args)
        return str(resp.content) if resp.content else "工具执行无输出"

    async def chat_base(self, messages: List[Dict[str, Any]]) -> Any:
        """一直循环到模型不再要求调用工具为止。"""
        response = self.client.chat.completions.create(
            model=self.model, messages=messages, tools=self.all_tools)
        while response.choices[0].finish_reason == "tool_calls":
            # 把这轮 assistant 消息原样回填，再逐个补上工具回执
            messages.append(response.choices[0].message.model_dump())
            for call in response.choices[0].message.tool_calls:
                result = await self._call_mcp_tool(
                    call.function.name, json.loads(call.function.arguments))
                messages.append({
                    "role": "tool",
                    "content": result,
                    "tool_call_id": call.id,
                })
            response = self.client.chat.completions.create(
                model=self.model, messages=messages, tools=self.all_tools)
        return response

    async def chat_loop(self) -> None:
        print("MCP 客户端已启动，输入 quit 退出。")
        messages: List[Dict[str, Any]] = []
        while True:
            query = input("\n你: ").strip()
            if query.lower() == "quit":
                break
            try:
                messages.append({"role": "user", "content": query})
                messages = messages[-20:]        # 只留最近 20 条，防止上下文撑爆
                response = await self.chat_base(messages)
                messages.append(response.choices[0].message.model_dump())
                print(f"\nAI: {response.choices[0].message.content}")
            except Exception as e:
                print(f"\n调用过程出错: {e}")

    async def cleanup(self) -> None:
        for server in self.servers.values():
            await server.cleanup()


async def main() -> None:
    config = Configuration()
    servers_config = config.load_config("servers_config.json")
    client = MultiServerMCPClient()
    try:
        await client.connect_to_servers(servers_config)
        await client.chat_loop()
    finally:
        # 给子进程一点时间收尾，再统一释放
        await asyncio.sleep(0.1)
        await client.cleanup()


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

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["weather_server.py"]
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "TODO-允许访问的目录"]
    },
    "git": {
      "command": "uvx",
      "args": ["mcp-server-git", "--repository", "TODO-本地仓库路径"]
    }
  }
}

# 复制为 .env 后填值，.env 必须加进 .gitignore
# 大模型服务的 API key
LLM_API_KEY=
# 大模型服务的兼容 OpenAI 接口地址
BASE_URL=
# 使用的模型名
MODEL=
# OpenWeather 的 API key，weather_server.py 读它
OPENWEATHER_API_KEY=