模型上下文协议MCP

      MCP(Model Context Protocol) Anthropic推出的一种开放协议，旨在统一LLM应用于外部数据源之间的通讯协议使之无缝集成，MCP提供了标准化协议使得LLM与所需要的上下文无缝衔接。使用MCP可以插件式为LLM的集成各种外部数据源。

MCP概念

      上图为MCP官方所描述的MCP架构图，MCP Hosts本身也是一个MCP Client，MCP Server端与各种类型外部数据源集成，一个Server可对应一个或多个外部数据源，数据源可以说本地数据或是网络数据。Client与Server建立一对一的连接，其通过MCP协议与Server端进行通讯获取Server所提供的各类数据。
      host：一个包含MCP Client的应用，可以是Web、App、或其他类型的程序等。
      MCP Client：使用MCP协议与Server建立一对一连接。
      MCP Server：连接内部、外部、网络资源，使用MCP协议对外提供服务。
      Local：内部资源
      Remote：外部/网络资源

MCP Server

      根据MCP协议定义，Server可以提供三种类型的标准能力，Resources、Tools、Prompts，每个Server可同时提供者三种类型能力或其中一种。
      Resources：资源，类似于文件数据读取，可以是文件资源或是API响应返回的内容。
      Tools：工具，第三方服务、功能函数，通过此可控制LLM可调用哪些函数。
      Prompts：提示词，为用户预先定义好的完成特定任务的模板。

通讯机制

      MCP定义了Client与Server进行通讯的协议与消息格式，其支持两种类型通讯机制：标准输入输出通讯、基于SSE的HTTP通讯，分别对应着本地与远程通讯。Client与Server间使用JSON-RPC 2.0格式进行消息传输。
      本地通讯：使用了stdio传输数据，具体流程Client启动Server程序作为子进程，其消息通讯是通过stdin/stdout进行的，消息格式为JSON-RPC 2.0。
      远程通讯：Client与Server可以部署在任何地方，Client使用SSE与Server进行通讯，消息的格式为JSON-RPC 2.0，Server定义了/see与/messages接口用于推送与接收数据。
      MCP定义了三类消息类型：Requests、Results、Errors、Notifications。

代码示例

      按照MCP协议的定义其是客户端-服务端模型，客户端请求使用服务端提供的工具、资源、提示词。这里会先编写Server端Demo然后再编写Client端，使用Client调用Server端。

Server

      在下面Server Demo中定义了Tool、Reource、Prompt各一个。

# -*- coding: utf-8 -*-
from codecs import encode
import logging
from typing import Any
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
from pydantic import AnyUrlserver = Server("demo")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:"""提示模版定义"""return [types.Prompt(name="example-prompt",description="An example prompt template",arguments=[types.PromptArgument(name="arg1",description="Example argument",required=True)])]@server.get_prompt()
async def handle_get_prompt(name: str,arguments: dict[str, str] | None
) -> types.GetPromptResult:"""提示模板处理"""if name != "example-prompt":raise ValueError(f"Unknown prompt: {name}")return types.GetPromptResult(description="Example prompt",messages=[types.PromptMessage(role="user",content=types.TextContent(type="text",text="Example prompt text"))])@server.list_resources()
async def list_resources() -> list[types.Resource]:"""资源定义"""test='test.txt'return [types.Resource(uri=AnyUrl(f"file:///{test}.txt"),name=test,description=f"A sample text resource named {test}",mimeType="text/plain",)# for name in SAMPLE_RESOURCES.keys()]@server.read_resource()
async def read_resource(uri: AnyUrl) -> str | bytes:assert uri.path is not Nonename = uri.path.replace(".txt", "").lstrip("/")if name not in SAMPLE_RESOURCES:raise ValueError(f"Unknown resource: {uri}")return SAMPLE_RESOURCES[name]@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:"""工具定义.每个工具都使用JSON Schema验证指定其参数."""return [types.Tool(name="demo-tool",description="Get data tool for a param",inputSchema={"type": "object","properties": {"param": {"type": "string","description": "url",},},"required": ["param"],},)]@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None
) -> list[Any]:logging.info(name)"""处理工具调用"""if not arguments:raise ValueError("Missing arguments")if name == "demo-tool":param = arguments.get("param")if not param:raise ValueError("Missing state parameter")param = param.upper()return [types.TextContent(type="text",text=f"text:{param}")]else:raise ValueError(f"Unknown tool: {name}")async def main():from anyio.streams.text import TextStream# Run the server using stdin/stdout streamsasync with mcp.server.stdio.stdio_server() as (read_stream, write_stream):await server.run(read_stream,write_stream,InitializationOptions(server_name="demo-server",server_version="0.1.0",capabilities=server.get_capabilities(notification_options=NotificationOptions(),experimental_capabilities={},),),)if __name__ == "__main__":
asyncio.run(main())

Client

      Client的实现，主要是与需要与Server建立连接，并调用Server所提供的各类资源。需要与Server所支持的方式建立连接，如其是stdio模式就需要建立stdio模式连接否则需要建立SSE连接。

async def demo(config):server_params = StdioServerParameters(command=shutil.which("npx") if config['command'] == "npx" else config['command'],args=config['args'],env={**os.environ, **config['env']} if config.get('env') else None)try:print('demo')stdio_context = stdio_client(server_params)read, write = await stdio_context.__aenter__()session = ClientSession(read, write)await session.__aenter__()capabilities = await session.initialize()tools_response =await session.list_tools()resources=await session.list_resources()tools = []for item in tools_response:if isinstance(item, tuple) and item[0] == 'tools':for tool in item[1]:tools.append(tool)params={}params[tools[0].inputSchema['required'][0]]='测试参数';result =await session.call_tool(tools[0].name,params)print(result)await stdio_context.__aexit__(None, None, None)except Exception as e:print(e)if __name__=="__main__":asyncio.run(demo({'command':'python','args':['server_demo.py']}))

MCP Inspector

      MCP提供了MCP Inspector 用于调试与测试MCP Server，使得Server开发更加方便。
      使用nodejs启动inspector程序，启动后在浏览器访问http://localhost:5173即可。可以在中查看MCP Server所提供的各种服务Tools、Resources、Prompts等，以及调用Tools服务。

npx @modelcontextprotocol/inspector

      使用MCP的意义主要是统一LLM访问外部资源的方式，可以限制LLM可使用哪些服务调用，第三方也可以定义各种类型的MCP Server提供给其他任何人使用。但Anthropic作为OpenAI的对手OpenAI跟不跟是MCP能否成为LLM上下文标准的关键。

参考资料：
MCP