MCP 解决什么问题

在 Agent 系统中，让Agent做出重要的决定的一个关键点是构造好的 /chat/completions 请求参数，参数中：

  1. messages 参数：给出对话历史，历史中包含的背景信息质量越高，给出的答案就会更准确；
  2. tools 参数：给出模型可用的工具，用于获取动态信息或执行某些动作。

这两个参数都可以认为是我们给模型的上下文信息。（context）

message 的内容和问题

messages 是一个数组，数组中的 content 的值是 user 给出的信息，或者其他角色的历史输出信息。如果是 user 给出的信息，则一般的做法是用一个 prompt template来组织基本的 prompt 结构，并将一些动态的信息通过其他方式检索到，补充道 prompt template中的占位符，从而形成最终输入模型的 prompt。

要得到最好的解，不同的问题需要不同的 prompt template。这是一个难点

tools 的内容和问题

/chat/completions 接受的 tool 是有一定的 schema 规定，而一般的工具（如RESTful API）最多有一个 openapi spec来规范，要扩展模型的能力边界，就要接入更多的工具。

而，工具是形形色色的，且验证方式可能有不同，接入也将是一个体力活。

MCP是解决上述问题的一种规范

MCP 是 client / server 架构，server 通过 api 提供 prompt template、tools、resources，client 则融合进调用大模型的客户端，与server建立连接，可动态、实时获取上下文信息。1

resources

Resource 指由Server提供的可由client直接访问的各类数据或内容，可以包含文本、文件内容、数据库记录、API返回的数据片段或其他结构化信息。Resource主要关注的是内容数据的提供，而不是执行操作或指导交互。

由上所述，因为resource可以是API返回的数据片段，好像和 tools 混在一起了？

其实不是：

  • resource 提供的是静态或现成的信息数据，是供参考和上下文补充的内容。是只读的，不会对系统状态产生直接改变，更多地是被“消费”来提升回答准确性。
  • tools 则提供的是可执行的操作接口，允许模型主动请求执行任务、产生更改或获取动态结果。

C/S 架构

Client

可以连多个 Server，嵌入client的应用负责：

  • 连接并管理servers
  • 汇聚来自多个servers的context

一个例子

使用 MCP 协议实现一个针对 PostgreSQL 数据库的服务端。在这个例子中，我们采用 JSON-RPC 2.0 格式作为通信机制，通过标准输入/输出进行请求和响应传递。

基本思路：

  1. 定义一个 MCP Server，它支持一个（或多个）方法，比如 “read_resource”，专门用于根据请求参数（例如 SQL 查询或资源 URI）从 PostgreSQL 数据库中获取数据。
  2. 使用 Python 的数据库库（例如 psycopg2）连接 PostgreSQL，并执行查询，将查询结果封装成符合 JSON-RPC 格式的响应返回。
  3. 将请求处理循环写入一个主函数中，按照 MCP 协议的格式读取和写入消息。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
#!/usr/bin/env python3
import json
import sys
import psycopg2
import urllib.parse

# 根据传入的查询字符串执行 PostgreSQL 查询，并返回结果
def query_postgresql(query_str):
    # 从环境变量或配置文件中读取数据库连接字符串
    # 示例中直接写明连接参数，仅供演示；实际环境建议放到配置文件或环境变量中
    conn_str = "dbname=test user=postgres password=yourpassword host=localhost port=5432"
    try:
        conn = psycopg2.connect(conn_str)
        cur = conn.cursor()
        cur.execute(query_str)
        # fetchall 获取所有查询结果，你也可以根据需求使用 fetchone/fetchmany
        rows = cur.fetchall()
        cur.close()
        conn.close()
        return rows
    except Exception as e:
        # 捕获异常并返回错误信息
        return {"error": str(e)}

# 处理 JSON-RPC 请求
def handle_request(request):
    method = request.get("method")
    params = request.get("params", {})
    request_id = request.get("id")
    
    if method == "read_resource":
        # 这里我们约定使用 "read_resource" 方法提供 PostgreSQL 数据查询功能
        # 可以约定传入的参数中包含SQL查询语句，或者解析 URI 中的 query 参数
        # 例如：{"uri": "postgresql://?query=SELECT+*+FROM+mytable"}
        uri = params.get("uri")
        query = params.get("query")  # 也可以直接通过 query 传入查询语句
        
        if uri:
            # 如果以 URI 方式传入，可解析其中的 query 参数
            parsed = urllib.parse.urlparse(uri)
            query_params = urllib.parse.parse_qs(parsed.query)
            query_list = query_params.get("query")
            if query_list:
                query = query_list[0]
        
        if not query:
            return {
                "jsonrpc": "2.0",
                "error": {"code": -32602, "message": "缺少查询参数"},
                "id": request_id,
            }
        
        # 执行 SQL 查询
        result = query_postgresql(query)
        # 构造响应
        return {"jsonrpc": "2.0", "result": result, "id": request_id}
    
    else:
        # 方法未定义
        return {
            "jsonrpc": "2.0",
            "error": {"code": -32601, "message": f"未找到方法: {method}"},
            "id": request_id,
        }

# 主循环：通过标准输入/输出通信
def main():
    while True:
        # 读取一行输入（JSON-RPC 请求）
        raw_line = sys.stdin.readline().strip()
        if not raw_line:
            break
        try:
            request = json.loads(raw_line)
        except Exception as e:
            # 如果解析失败则输出错误
            error_response = {
                "jsonrpc": "2.0",
                "error": {"code": -32700, "message": f"无效的JSON: {str(e)}"},
                "id": None,
            }
            sys.stdout.write(json.dumps(error_response) + "\n")
            sys.stdout.flush()
            continue

        response = handle_request(request)
        sys.stdout.write(json.dumps(response) + "\n")
        sys.stdout.flush()

if __name__ == "__main__":
    main()

测试

1
echo '{"jsonrpc": "2.0", "method": "read_resource", "params": {"query": "SELECT * FROM mytable"}, "id": 1}' | python mcp_se