Jean's Blog

一个专注软件测试开发技术的个人博客

0%

MCP服务开发的核心概念和使用

发表于 分类于
学习MCP学习笔记第三讲

MCP 服务

模型上下文协议（MCP）允许你构建服务器，以安全、标准化的方式向LLM应用暴露数据和功能。可以把它看作是一个专门为大型语言模型交互设计的Web API。MCP服务器可：

  • 通过资源暴露数据（可以把它们想象成GET端点;它们用来将信息加载到LLM的上下文中）
  • 通过工具提供功能（有点像POST端点;它们用于执行代码或产生副作用）
  • 通过提示（用于大型语言模型交互的可复用模板）定义交互模式
  • mcp 库的文档：https://github.com/modelcontextprotocol/python-sdk

定义MCP的工具（Tools）

工具允许LLM通过您的服务器执行操作。与资源不同，工具需要执行计算并具有副作用。

MCP工具的定义规范：

  1. 必须要定义工具服务的，文档注释（告诉大模型或Agent，该工具的作用是什么）
  2. 工具必须定义参数及声明的参数类型（告诉大模型或Agent，该工具传什么样的参数）
  3. 对于工具返回的数据，类型和数据结构要声明（告诉大模型或Agent，该工具执行完之后的结果是什么样的，方便后面去进行解析和使用）

示例：

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
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="Tool Example")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算数字相加"""
    return a + b
  
@mcp.tool()
def http_request(
    url: str,
    method: str = "GET",
    headers: Optional[Dict[str, str]] = None,
    params: Optional[Dict[str, Any]] = None,
    json_body: Optional[Dict[str, Any]] = None,
    form_body: Optional[Dict[str, Any]] = None,
    timeout: int = 10
):
    """
    通用 HTTP 请求工具，可用于所有网络请求。

    参数：
    - url: 请求地址
    - method: HTTP 方法，如 GET / POST / PUT / DELETE
    - headers: 字典型请求头
    - params: 查询参数 (?key=value)
    - json_body: JSON body，例如 POST JSON
    - form_body: 表单 body，例如 POST x-www-form-urlencoded
    - timeout: 超时时间（秒）
    """

    method = method.upper()

    try:
        resp = requests.request(
            method=method,
            url=url,
            headers=headers,
            params=params,
            json=json_body,
            data=form_body,
            timeout=timeout,
        )
        # 根据 Content-Type 决定是否尝试 JSON 解析
        content_type = resp.headers.get("Content-Type", "")
        if "application/json" in content_type:
            try:
                retrun resp.json()
            except Exception:
                retrun resp.text
        else:
            retrun resp.text

定义MCP的资源（Resources）

资源是你向大型语言模型展示数据的方式。它们类似于REST API中的GET端点——它们提供数据，但不应该执行大量计算或带来副作用。

示例

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
from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="Resource Example")

# =====================================================
# 资源 1：读取测试报告文件
# =====================================================
@mcp.resource("report://{filename}")
def read_test_report(filename: str) -> str:
    """读取测试报告内容（如 test-report.log / junit.xml / coverage.txt）"""
    try:
        with open(filename, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"测试报告文件 {filename} 不存在"
    except Exception as e:
        return f"读取测试报告失败: {str(e)}"


# =====================================================
# 资源 2：读取测试用例文件（支持 YAML / JSON）
# =====================================================
@mcp.resource("testcase://{filename}")
def read_test_case_file(filename: str) -> dict:
    """
    读取测试用例文件：
    - JSON 格式测试用例
    - YAML 格式测试用例
    """
    try:
        with open(filename, "r", encoding="utf-8") as f:
            content = f.read()
        # JSON
        if filename.endswith(".json"):
            return json.loads(content)
        # YAML
        if filename.endswith((".yaml", ".yml")):
            return yaml.safe_load(content)
        return {"error": "不支持的测试用例文件格式"}
    except FileNotFoundError:
        return {"error": f"文件 {filename} 不存在"}
    except Exception as e:
        return {"error": f"解析测试用例文件失败: {str(e)}"}

# =====================================================
# 资源 3：获取应用程序设置
# =====================================================
@mcp.resource("config://settings")
def get_settings() -> str:
    """获取应用程序设置。"""
    return """{
  "theme": "dark",
  "language": "en",
  "debug": false
}"""
  
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

资源加载

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
# @Time：2026/1/4 14:05
# @Author：jinglv
import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


async def create_mcp_stream_client():
    """创建MCP客户端"""
    # 创建 MCP 客户端
    client = MultiServerMCPClient(
        {
            "resource-example": {
                "url": "http://127.0.0.1:8000/mcp",
                "transport": "streamable_http",
            }
        }
    )

    # 异步获取工具列表（通过 MCP 协议动态加载工具）
    # tools = await client.get_tools()
    # print("工具列表：", tools)

    # 加载mcp服务中的资源resource
    # 创建一个mcp的会话对象
    async with client.session("resource-example") as session:
        # 通过mcp会话对象去获取mcp服务中的资源列表
        resources = await session.list_resource_templates()
        print("resources:", resources.resourceTemplates)

    # agent = create_agent(
    #     model_client,
    #     tools=tools,
    #     system_prompt="你是一个计算专家智能助手，可以调用工具进行计算。"
    # )
    # # 3.异步运行智能体
    # res = await agent.ainvoke({"messages": [{"role": "user", "content": "100+100=？"}]})
    # return res


if __name__ == '__main__':
    res = asyncio.run(create_mcp_stream_client())
    # print(res)

工具结构化输出

默认情况下，工具将返回结构化结果，如果其返回类型注释是兼容的。否则，它们将返回非结构化结果。

结构化输出支持以下返回类型：

  • Pydantic模型（BaseModel子类）
  • 类型字典
  • 数据类和其他具有类型提示的类
  • dict[str, T]（其中T是任何JSON可序列化类型）
  • 原始类型（str、int、float、bool、bytes、None）- 包装在{“result”: value}
  • 泛型类型（列表、元组、联合、可选等）- 包装在{“result”: value}
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
from typing import TypedDict
from pydantic import BaseModel, Field

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Structured Output Example")

# 使用Pydantic模型获得丰富的结构化数据
class WeatherData(BaseModel):
    """天气信息结构。"""

    temperature: float = Field(description="摄氏温度")
    humidity: float = Field(description="湿度百分比")
    condition: str
    wind_speed: float

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """获取城市天气 - 返回结构化数据。"""
    # 模拟天气数据
    return WeatherData(
        temperature=72.5,
        humidity=45.0,
        condition="sunny",
        wind_speed=5.2,
    )

# 使用TypedDict获得更简单的结构
class LocationInfo(TypedDict):
    latitude: float
    longitude: float
    name: str

@mcp.tool()
def get_location(address: str) -> LocationInfo:
    """获取位置坐标"""
    return LocationInfo(latitude=51.5074, longitude=-0.1278, name="英国伦敦")

# 使用dict[str, Any]获得灵活的模式
@mcp.tool()
def get_statistics(data_type: str) -> dict[str, float]:
    """获取各种统计数据"""
    return {"mean": 42.5, "median": 40.0, "std_dev": 5.2}

# 具有类型提示的普通类也可以用于结构化输出
class UserProfile:
    name: str
    age: int
    email: str | None = None

    def __init__(self, name: str, age: int, email: str | None = None):
        self.name = name
        self.age = age
        self.email = email

@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    """获取用户资料 - 返回结构化数据"""
    return UserProfile(name="Alice", age=30, email="alice@example.com")

# 没有类型提示的类无法用于结构化输出
class UntypedConfig:
    def __init__(self, setting1, setting2):
        self.setting1 = setting1
        self.setting2 = setting2

@mcp.tool()
def get_config() -> UntypedConfig:
    """这返回非结构化输出 - 没有生成模式"""
    return UntypedConfig("value1", "value2")

# 列表和其他类型会自动包装
@mcp.tool()
def list_cities() -> list[str]:
    """获取城市列表"""
    return ["伦敦", "巴黎", "东京"]
    # 返回: {"result": ["伦敦", "巴黎", "东京"]}

@mcp.tool()
def get_temperature(city: str) -> float:
    """获取温度作为简单浮点数"""
    return 22.5
    # 返回: {"result": 22.5}
    
class TestCase(BaseModel):
    """单个测试用例结构"""
    id: str = Field(description="测试用例ID")
    title: str = Field(description="测试标题")
    steps: list[str] = Field(description="执行步骤")
    expected: str = Field(description="预期结果")
    priority: str = Field(description="优先级：P0, P1, P2")
    tags: list[str] = Field(description="测试标签")

@mcp.tool()
def generate_test_case(api_name: str) -> TestCase:
    """根据 API 名称自动生成一个测试用例（Pydantic输出）"""
    return TestCase(
        id="TC-API-001",
        title=f"{api_name} 接口基础功能验证",
        steps=[
            f"调用接口 {api_name}",
            "检查响应状态码是否为 200",
            "验证响应体结构是否正确"
        ],
        expected="接口返回 200 且响应体字段符合规范",
        priority="P1",
        tags=["api", "smoke", api_name]
    )

定义MCP的提示词（Prompts）

提示是可重用的模板，可帮助LLM有效地与您的服务器交互。

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
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts import base

mcp = FastMCP(name="Prompt Example")

@mcp.prompt(title="代码审查")
def review_code(code: str) -> str:
    return f"请审查此代码:\n\n{code}"

@mcp.prompt(title="调试助手")
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("我看到了这个错误:"),
        base.UserMessage(error),
        base.AssistantMessage("我会帮你调试。你试过什么方法？"),
    ]
    
@mcp.prompt("测试用例生成提示词")
def test_case_prompt(requirement: str, level: str = "normal") -> str:
    """
    根据需求生成测试用例的提示词模板。
    requirement: 业务 / 功能需求描述
    level: 用例粒度等级（smoke / normal / detailed）
    """
    return f"""
        你是一名专业的测试工程师，请根据以下需求生成测试用例。
        
        【需求描述】
        {requirement}
        
        【测试粒度】
        {level}
        
        【生成要求】
        1. 给出 3~5 条测试用例。
        2. 用例必须包含：标题、前置条件、步骤、预期结果。
        3. 步骤应具体、可执行。
        4. 如果需求存在边界情况或异常分支，请一并生成测试。
        5. 输出格式必须为 JSON 数组，每个元素包含：
           - id：唯一ID
           - title：测试标题
           - preconditions：数组
           - steps：数组
           - expected：字符串
           - tags：如 ["api", "regression", "edge"]
        6. 禁止生成与需求无关的内容。
        
        请严格按照以上规则生成输出。
        """
 
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

模版信息加载

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
# @Time：2026/1/4 14:13
# @Author：jinglv
import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


async def create_mcp_stream_client():
    """创建MCP客户端"""
    # 创建 MCP 客户端
    client = MultiServerMCPClient(
        {
            "prompts-example": {
                "url": "http://127.0.0.1:8000/mcp",
                "transport": "streamable_http",
            }
        }
    )

    # 异步获取工具列表（通过 MCP 协议动态加载工具）
    # tools = await client.get_tools()
    # print("工具列表：", tools)

    # 加载mcp服务中的资源resource
    # 创建一个mcp的会话对象
    async with client.session("prompts-example") as session:
        # 通过mcp会话对象去获取mcp服务中的资源列表
        resources = await session.list_prompts()
        print("resources:", resources.prompts)


if __name__ == '__main__':
    res = asyncio.run(create_mcp_stream_client())

定义MCP的上下文（Context）

Context对象使您的工具和资源能够访问MCP功能。

上下文属性与方法

上下文对象提供以下功能：

  • ctx.request_id- 当前请求的唯一ID
  • ctx.client_id- 客户ID （如有）
  • ctx.fastmcp- 访问FastMCP服务器实例（参见FastMCP属性）)
  • ctx.session- 访问底层会话以实现高级通信（参见会话属性和方法)
  • ctx.request_context- 访问请求特定数据和生命周期资源（参见请求上下文属性）)
  • await ctx.debug(message)- 发送调试日志消息
  • await ctx.info(message)- 发送信息日志消息
  • await ctx.warning(message)- 发送警告日志消息
  • await ctx.error(message)- 发送错误日志消息
  • await ctx.log(level, message, logger_name=None)- 发送带有自定义电平的日志
  • await ctx.report_progress(progress, total=None, message=None)- 报告运营进展
  • await ctx.read_resource(uri)- 阅读URI的资源
  • await ctx.elicit(message, schema)- 向用户请求额外信息并进行验证
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from mcp.server.fastmcp import Context, FastMCP

mcp = FastMCP(name="Progress Example")

@mcp.tool()
async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
    """执行带进度更新的任务。"""
    await ctx.info(f"开始: {task_name}")

    for i in range(steps):
        progress = (i + 1) / steps
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"步骤 {i + 1}/{steps}",
        )
        await ctx.debug(f"完成步骤 {i + 1}")

    return f"任务 '{task_name}' 完成"

MCP 添加认证

认证架构：

  • 授权服务器（AS）：处理OAuth流、用户身份验证和令牌颁发
  • 资源服务器（RS）：验证令牌并为受保护资源提供服务的MCP服务器
  • 客户端：获取令牌，提交并将其与MCP服务器进行验证

MCP服务器可以通过提供协议的TokenVerifier实现来使用身份验证。

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
from pydantic import AnyHttpUrl

from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import FastMCP

class SimpleTokenVerifier(TokenVerifier):
    """用于演示的简单令牌验证器。"""

    async def verify_token(self, token: str) -> AccessToken | None:
        pass  # 这里您需要实现实际的令牌验证

# 创建FastMCP实例作为资源服务器
mcp = FastMCP(
    "Weather Service",
    # 用于身份验证的令牌验证器
    token_verifier=SimpleTokenVerifier(),
    # 认证设置
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.example.com"),  # 授权服务器URL
        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # 此服务器的URL
        required_scopes=["user"],
    ),
)

@mcp.tool()
async def get_weather(city: str = "London") -> dict[str, str]:
    """获取城市天气数据"""
    return {
        "city": city,
        "temperature": "22",
        "condition": "Partly cloudy",
        "humidity": "65%",
    }

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

带有认证的MCP client示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "streamable_http",
            "url": "http://localhost:8000/mcp",
            # 设置请求头，配置用于身份验证的token
            "headers": {
                "Authorization": "Bearer YOUR_TOKEN",
                "X-Custom-Header": "custom-value"
            },
        }
    }
)

图片处理工具

FastMCP 提供了一个自动处理图像数据的类，可以用来对图片进行处理，安装依赖：pip install pillow

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
from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage
import io

mcp = FastMCP("ImageProcessor")

@mcp.tool()
def create_thumbnail(image_path: str, size: int = 100) -> Image:
    """创建正方形缩略图"""
    img = PILImage.open(image_path)
    img.thumbnail((size, size))
    
    # 转换为字节
    byte_arr = io.BytesIO()
    img.save(byte_arr, format='PNG')
    
    return Image(data=byte_arr.getvalue(), format="png")

@mcp.tool()
def resize_image(image_path: str, width: int, height: int) -> Image:
    """调整图片到指定尺寸（可能变形）"""
    img = PILImage.open(image_path)
    resized = img.resize((width, height))
    
    byte_arr = io.BytesIO()
    resized.save(byte_arr, format='PNG')
    
    return Image(data=byte_arr.getvalue(), format="png")