生态系统：MCP 提供了丰富的插件库，使您的 AI 应用可以直接利用。所有的服务提供商都可以按MCP协议进行接入，开者可以直接使用。可以预见，这个生态会越来越庞大，生态越庞大，AI的能力越强。
兼容性：大家只要遵循一套协议，都可以实现“万物互联”。不限于特定的 AI 模型，任何支持 MCP 的模型都能灵活切换使用。

客户端（Claude Desktop / Cursor）将问题发送给 LLM。
LLM 分析可用的工具，并决定使用哪一个（或多个）。
客户端通过 MCP Server 执行所选的工具。
工具的执行结果被送回给 LLM。
LLM 结合执行结果，归纳总结后生成自然语言展示给用户！

... # 省略了无关的代码
 async def start(self):
     # 初始化所有的 mcp server
     for server in self.servers:
         await server.initialize()

     # 获取所有的 tools 命名为 all_tools
     all_tools = []
     for server in self.servers:
         tools = await server.list_tools()
         all_tools.extend(tools)


     # 将所有的 tools 的功能描述格式化成字符串供 LLM 使用
     # tool.format_for_llm() 我放到了这段代码最后，方便阅读。
     tools_description = "\n".join(
         [tool.format_for_llm() for tool in all_tools]
     )


     # 询问 LLM（Claude） 应该使用哪些工具。
     system_message = (
         "You are a helpful assistant with access to these tools:\n\n"
         f"{tools_description}\n"
         "Choose the appropriate tool based on the user's question. "
         "If no tool is needed, reply directly.\n\n"
         "IMPORTANT: When you need to use a tool, you must ONLY respond with "
         "the exact JSON object format below, nothing else:\n"
         "{\n"
         '    "tool": "tool-name",\n'
         '    "arguments": {\n'
         '        "argument-name": "value"\n'
         "    }\n"
         "}\n\n"
         "After receiving a tool's response:\n"
         "1. Transform the raw data into a natural, conversational response\n"
         "2. Keep responses concise but informative\n"
         "3. Focus on the most relevant information\n"
         "4. Use appropriate context from the user's question\n"
         "5. Avoid simply repeating the raw data\n\n"
         "Please use only the tools that are explicitly defined above."
     )
     messages = [{"role": "system", "content": system_message}]

     while True:
         # Final... 假设这里已经处理了用户消息输入.
         messages.append({"role": "user", "content": user_input})

         # 将 system_message 和用户消息输入一起发送给 LLM
         llm_response = self.llm_client.get_response(messages)

     ... # 后面和确定使用哪些工具无关


 class Tool:
     """Represents a tool with its properties and formatting."""

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

     # 把工具的名字 / 工具的用途（description）和工具所需要的参数（args_desc）转化为文本
     def format_for_llm(self) -> str:
         """Format tool information for LLM.

         Returns:
             A formatted string describing the tool.
         """
         args_desc = []
         if "properties" in self.input_schema:
             for param_name, param_info in self.input_schema["properties"].items():
                 arg_desc = (
                     f"- {param_name}: {param_info.get('description', 'No description')}"
                 )
                 if param_name in self.input_schema.get("required", []):
                     arg_desc += " (required)"
                 args_desc.append(arg_desc)

         return f"""
 Tool: {self.name}
 Description: {self.description}
 Arguments:
 {chr(10).join(args_desc)}
 """

@classmethod
 def from_function(
     cls,
     fn: Callable,
     name: str | None = None,
     description: str | None = None,
     context_kwarg: str | None = None,
 ) -> "Tool":
     """Create a Tool from a function."""
     func_name = name or fn.__name__ # 获取函数名

     if func_name == "<lambda>":
         raise ValueError("You must provide a name for lambda functions")

     func_doc = description or fn.__doc__ or "" # 获取函数 docstring
     is_async = inspect.iscoroutinefunction(fn)

     ... # 更多请参考原始代码...

无需工具时：模型直接生成自然语言回复。
需要工具时：模型输出结构化 JSON 格式的工具调用请求。

... # 省略无关的代码
 async def start(self):
     ... # 上面已经介绍过了，模型如何选择工具

     while True:
         # 假设这里已经处理了用户消息输入.
         messages.append({"role": "user", "content": user_input})

         # 获取 LLM 的输出
         llm_response = self.llm_client.get_response(messages)

         # 处理 LLM 的输出（如果有 tool call 则执行对应的工具）
         result = await self.process_llm_response(llm_response)

         # 如果 result 与 llm_response 不同，说明执行了 tool call （有额外信息了）
         # 则将 tool call 的结果重新发送给 LLM 进行处理。
         if result != llm_response:
             messages.append({"role": "assistant", "content": llm_response})
             messages.append({"role": "system", "content": result})
             final_response = self.llm_client.get_response(messages)
             logging.info("\nFinal response: %s", final_response)
             messages.append(
                 {"role": "assistant", "content": final_response}
             )
         # 否则代表没有执行 tool call，则直接将 LLM 的输出返回给用户。
         else:
             messages.append({"role": "assistant", "content": llm_response})

访问 https://modelcontextprotocol.io/llms-full.txt 并复制完整的文档文本。（实测这个太长了，可以忽略）
导航到 MCP TypeScript SDK 或 Python SDK Github 项目中并复制相关内容。
把这些作为 prompt 输入到你的 chat 对话中（作为 context）。

你的服务器会开放哪些资源
它会提供哪些工具
它应该给出哪些引导或建议
它需要跟哪些外部系统互动

... （这里是已经引入的 domain knowledge）


打造一个 MCP 服务器，它能够：


- 连接到我公司的 PostgreSQL 数据库
- 将表格结构作为资源开放出来
- 提供运行只读 SQL 查询的工具
- 包含常见数据分析任务的引导

# 安装 uv
curl -LsSf https://astral.sh/uv/install.sh | sh


# 创建项目目录
uv init txt_counter
cd txt_counter


# 设置 Python 3.10+ 环境
echo "3.11" > .python-version


# 创建虚拟环境并激活
uv venv
source .venv/bin/activate


# Install dependencies
uv add "mcp[cli]" httpx


# Create our server file
touch txt_counter.py

"""
... （这里是已经引入的 domain knowledge）
"""


打造一个 MCP 服务器，它能够：
- 功能：
    - 统计当前桌面上的 txt 文件数量
    - 获取对应文件的名字


要求：
- 不需要给出 prompt 和 resource 相关代码。
- 你可以假设我的桌面路径为 /Users/{username}/Desktop

import os
from pathlib import Path
from mcp.server.fastmcp import FastMCP


# 创建 MCP Server
mcp = FastMCP("桌面 TXT 文件统计器")


@mcp.tool()
def count_desktop_txt_files() -> int:
    """Count the number of .txt files on the desktop."""
    # Get the desktop path
    username = os.getenv("USER") or os.getenv("USERNAME")
    desktop_path = Path(f"/Users/{username}/Desktop")

    # Count .txt files
    txt_files = list(desktop_path.glob("*.txt"))
    return len(txt_files)


@mcp.tool()
def list_desktop_txt_files() -> str:
    """Get a list of all .txt filenames on the desktop."""
    # Get the desktop path
    username = os.getenv("USER") or os.getenv("USERNAME")
    desktop_path = Path(f"/Users/{username}/Desktop")

    # Get all .txt files
    txt_files = list(desktop_path.glob("*.txt"))

    # Return the filenames
    if not txt_files:
        return "No .txt files found on desktop."

    # Format the list of filenames
    file_list = "\n".join([f"- {file.name}" for file in txt_files])
    return f"Found {len(txt_files)} .txt files on desktop:\n{file_list}"


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run()

$ mcp dev txt_counter.py
Starting MCP inspector...
Proxy server listening on port 3000


MCP Inspector is up and running at http://localhost:5173

# 打开 claude_desktop_config.json (MacOS / Linux)
# 如果你用的是 cursor 或者 vim 请更换对应的命令
code ~/Library/Application\ Support/Claude/claude_desktop_config.json

{
    "mcpServers": {
      "txt_counter": {
        "command": "/opt/homebrew/bin/uv",
        "args": [
          "--directory",
          "/Users/yangfan/mcp/txt_counter",
          "run",
          "txt_counter.py"
        ]
      }
    }
  }

能推测我当前桌面上 txt 文件名的含义吗？

阅读原文

跳转微信打开

Fish AI Reader

Fish AI Reader

AI辅助创作，多种专业模板，深度分析，高质量内容生成。从观点提取到深度思考，FishAI为您提供全方位的创作支持。新版本引入自定义参数，让您的创作更加个性化和精准。

FishAI

鱼阅，AI 时代的下一个智能信息助手，助你摆脱信息焦虑

联系邮箱 441953276@qq.com