agentscope 源码分析（2）：MCP & Agent Skills

上一篇文章对 AgentScope 工具调用功能的核心类 Toolkit 类做了较为深入的分析，这篇文章我们继续分析 AgentScope 工具能力的另外几个特性，包括 MCP 工具、Agent Skills。

MCP 简介

MCP（Model Context Protocol，模型上下文协议）是一个旨在标准化 AI 模型与外部数据源和工具之间交互的开放协议。在上篇文章中，我们都是将自己代码中的 Python 函数注册为可被 LLM 调用的工具，而 MCP 协议则提供了一种通用、标准的方式，让AI模型能够连接和调用各种工具与数据，无论这些工具是如何实现的、部署在哪里。

在 MCP 出现之前，如果你想让 AI 访问你的 Google Drive、GitHub 或者本地数据库，每个开发者都要为不同的 AI 模型写一套不同的 连接器。而 MCP 则像是 AI 界的 USB 接口标准。只要数据源支持 MCP，任何支持 MCP 的 AI 客户端（如 Claude Desktop, Cursor）都能直接插上使用。

MCP 采用了典型的 客户端-服务器（Client-Server）架构：

• MCP Host (主机)：这是用户直接交互的 AI 应用程序，例如 Claude Desktop、Cursor 或者 自定义的 AI Agent。这个 AI 应用会与 LLM 通信，并发起可能的工具调用请求

• MCP Client (客户端)：Client 是 Host 内部的轻量级组件，负责与一个特定的 MCP Server 建立和维护一对一的连接，充当 Host 与 Server 之间的 翻译官，将 Host 的请求转发给 Server，并将 Server 的响应返回给Host

• MCP Server (服务端)：Server是MCP架构中的核心，它通过标准化接口，向客户端暴露三种主要能力

  • 资源 (Resources)：任何可供模型读取的数据，例如文件内容、数据库记录或API响应。每个资源由唯一的 URI 标识
  • 工具 (Tools)：模型可以执行的函数，工具通常可以改变状态或与外部系统交互
  • 提示 (Prompts)：预定义的提示词模板，可以接受动态参数，帮助用户更高效地完成特定任务

MCP 协议定义了标准化的传输方式，以确保不同组件间的可靠通信：

• STDIO：主要用于本地通信，此时 MCP Client 会将 MCP Server 作为子进程启动，然后通过标准输入（stdin）和标准输出（stdout）进行 JSON-RPC 消息的交换
• Streamable HTTP：用于远程通信（早期版本使用 SSE，现已演进），这使得 MCP Server 可以作为独立的网络服务运行，供远程客户端连接。客户端通过 HTTP POST 发送请求，并通过 Server-Sent Events (SSE) 接收服务端推送的消息

所有 MCP 消息都必须遵循 JSON-RPC 2.0 规范，定义了请求、响应和通知三种基本消息类型，确保了交互的结构化与一致性。

MCP 的一个核心优势是：开发者无需为每个模型单独编写工具调用代码，只需将工具实现为一次 MCP Server，任何支持 MCP 的 Agent 都能直接使用。这样工具的开发与 LLM/Agent 之间的集成就被解耦了。

有一点需要注意，MCP 并不是要取代 LLM 的 Function Call（Tool Call，工具调用），Function Call 定义了模型如何输出 JSON 来表达 我想调用这个函数，而 MCP 协议则定义了如何发现发现、连接、运行各种工具，因此 Agent 仍然还是要依赖 LLM 的 Function Call 能力来触发工具的调用，而 MCP 协议则标准化了工具的发现、调用过程。实际运行中，两者之间是协同工作的:

• 连接：你的 AI Agent 应用（Host）连接到一个 MCP Server（比如一个 GitHub 统计工具）。
• 发现：MCP Server 告诉客户端： 我有 get_repo_stars 这个工具
• 注入：客户端把这个工具的定义，通过 Function Call 的格式告诉 LLM
• 决策（Function Call 发生）：LLM 决定调用它，输出：{"name": "get_repo_stars", "args": {...}}
• 执行（MCP 工具调用发生）：客户端收到这个 JSON，通过 MCP 协议转发给对应的 Server 去执行
• 返回：Server 把结果传回 MCP 客户端，AI Agent 应用再通过 Tool Result 的形式喂给 LLM

所以要注意 LLM 的 Function Call 与 MCP 协议之间的联系与区别：

• Function Call 解决了 模型想干什么 的问题
• MCP 解决了 工具在哪、怎么连、怎么传数据 的问题

AgentScope 的 MCP 模块

src/agentscope/mcp 目录是 AgentScope 实现 MCP 工具调用功能的核心代码，每个文件的主要作用：

文件	类/功能
_client_base.py	MCPClientBase - 客户端抽象基类
_stateful_client_base.py	StatefulClientBase - 有状态客户端基类
_http_stateful_client.py	HttpStatefulClient - HTTP 有状态客户端
_http_stateless_client.py	HttpStatelessClient - HTTP 无状态客户端
_stdio_stateful_client.py	StdIOStatefulClient - StdIO 有状态客户端
_mcp_function.py	MCPToolFunction - 工具函数封装

当然 AgentScope 的 MCP 模块并没有从头开始实现 MCP 协议，而是对现有的 MCP Python SDK 进行封装和适配。总体来说，提供了以下几个接口：

功能	说明
连接管理	建立和维护与 MCP 服务器的连接
工具发现	列出服务器提供的可用工具
工具调用	将 MCP 工具封装为可直接调用的函数
结果转换	将 MCP 格式结果转为 AgentScope 格式
生命周期管理	管理有状态/无状态客户端的会话

MCPClientBase (客户端基类)

MCPClientBase 定义了 MCP Client 的抽象基类，其中：

• get_callable_function 是一个抽象方法，具体的 Client 实现类需要实现这个方法，用于根据工具名称获取一个可调用对象
• _convert_mcp_content_to_as_blocks 方法用于将 MCP 返回的结果类型转换为 AgentScope 自己定义的内容格式，例如 TextBlock、ImageBlock 等

class MCPClientBase:
    def __init__(self, name: str):
        self.name = name  # MCP 服务器唯一标识符

    @abstractmethod
    async def get_callable_function(self, func_name: str) -> Callable:
        """获取可调用的工具函数"""

    @staticmethod
    def _convert_mcp_content_to_as_blocks(mcp_content_blocks) -> list:
        """将 MCP 内容块转换为 AgentScope Block"""

StatefulClientBase（有状态客户端基类）

AgentScope 将 MCP client 分为 有状态 与 无状态 两种，他们的区别在于客户端是否会维持与 MCP 服务器的会话（session）：

• 有状态客户端：其生命周期内始终维持与 MCP 服务器的持久会话，需要开发者显式调用 connect() 和 close() 方法来管理会话的生命周期
• 无状态客户端只会在调用工具发生时建立会话，并在工具调用结束后立即销毁会话，是一种轻量化的使用方式

特性	有状态客户端	无状态客户端
会话管理	手动 connect()/close()	自动管理
性能	多次调用共享会话	每次调用新建会话
适用场景	需要状态保持（如浏览器）	简单工具调用
资源开销	低（连接复用）	高（每次新建连接）

StatefulClientBase 是有状态客户端的基类，管理持久会话，适用于需要维护状态的 MCP 服务器（如浏览器自动化）。

is_connected: bool      # 连接状态
client: Any             # MCP 客户端实例
session: ClientSession  # MCP 会话
stack: AsyncExitStack   # 资源管理栈
self._cached_tools = None # 缓存的已工具函数

async def connect(self) -> None:
    """建立连接，初始化会话"""
    context = await self.stack.enter_async_context(self.client)
    read_stream, write_stream = context, context
    self.session = ClientSession(read_stream, write_stream)
    await self.stack.enter_async_context(self.session)
    await self.session.initialize()
    self.is_connected = True

async def close(self) -> None:
    """关闭连接，释放资源"""
    await self.stack.aclose()

async def list_tools(self) -> List[mcp.types.Tool]:
    """列出服务器提供的所有工具"""
    res = await self.session.list_tools()
    self._cached_tools = res.tools

async def get_callable_function(self, func_name: str) -> MCPToolFunction:
    """获取指定名称的工具函数"""
    # 返回 MCPToolFunction 所表示的可调用对象，转换了 `mcp.types.Tool`

StatefulClientBase 使用 Python contextlib 的 AsyncExitStack 来管理异步资源的生命周期，包括与 MCP 服务器的 HTTP 连接和逻辑回话两种资源。

HttpStatefulClient（HTTP 有状态客户端）

HttpStatefulClient 继承自 StatefulClientBase，实现了基于 HTTP 协议的有状态 MCP 客户端。transport 可以是 streamable_http 或者 sse：

class HttpStatefulClient(StatefulClientBase):
    def __init__(
        self,
        name: str,
        transport: Literal["streamable_http", "sse"],  # 传输类型
        url: str,                                       # 服务器 URL
        headers: dict | None = None,                    # HTTP 头
        timeout: float = 30,                            # 请求超时
        sse_read_timeout: float = 300,                  # SSE 读取超时
    ):
        if transport == "streamable_http":
            self.client = streamablehttp_client(url=url, ...)
        else:  # sse
            self.client = sse_client(url=url, ...)

• streamable HTTP 是原生 HTTP 流传输机制，可以将任意数据（二进制、纯文本、JSON块）以 chunk 方式传输。服务器在 Header 中声明 Transfer-Encoding: chunked 开启流传输模式。

• SSE（Server-Sent Events）则是 HTTP 之上的构造出的特定传输格式，专门为 服务器推向客户端 设计的。通过 event、data 等固定格式的文本消息来传递数据

Content-Type: text/event-stream

event: user_message
data: {"text": "Hello"}

event: system_update
data: {"status": "processing"}

HttpStatelessClient (HTTP 无状态客户端)

每次工具调用创建新会话，适用于无状态 MCP 服务器：

class HttpStatelessClient(MCPClientBase):
    stateful: bool = False  # 标记为无状态

    def get_client(self) -> _AsyncGeneratorContextManager:
        """获取一次性客户端实例"""
        if self.transport == "sse":
            return sse_client(**self.client_config)
        return streamablehttp_client(**self.client_config)

    async def list_tools(self) -> List[mcp.types.Tool]:
        """临时连接获取工具列表"""
        async with self.get_client() as cli:
            async with ClientSession(...) as session:
                await session.initialize()
                return await session.list_tools()

StdIOStatefulClient (StdIO 有状态客户端)

基于 STDIO Transport 的 MCP 客户端，它总是有状态的，当调用 connect() 时，它将在本地启动 MCP 服务器然后通过标准输入/输出与 MCP 服务器通信。

class StdIOStatefulClient(StatefulClientBase):
    def __init__(......):
        self.client = stdio_client(
            StdioServerParameters(
                command=command,
                args=args or [],
                env=env,
                cwd=cwd,
                encoding=encoding,
                encoding_error_handler=encoding_error_handler,
            ),
        )

MCPToolFunction (MCP 工具函数封装)

上篇文章中说过，所有的工具函数最终都会通过 register_tool_function 注册到 Toolkit 中，这些工具函数可以是普通的 Python 函数，也可以是 MCP 工具函数。AgentScope 框架将 MCP 工具函数 用 MCPToolFunction 来表示，其核心目的是将 MCP 工具函数直接封装为一个 Callable 对象，之后 Toolkit 就可以不加区别地处理普通函数和 MCP 函数。

MCPToolFunction 实现了 __call__ 方法，因此它的对象就是一个可调用对象，而在 __call__ 方法中，它会发起 MCP 调用：

class MCPToolFunction:
    name: str           # 工具名称
    description: str    # 工具描述
    json_schema: dict   # 参数 JSON Schema

    ......

    async def __call__(self, **kwargs) -> ToolResponse | CallToolResult:
        """执行工具调用"""
        if self.client_gen:  # 无状态模式，总是新建连接和会话
            async with self.client_gen() as cli:
                async with ClientSession(...) as session:
                    res = await session.call_tool(self.name, arguments=kwargs)
        else:  # 有状态模式
            res = await self.session.call_tool(self.name, arguments=kwargs)

        # 是否将 MCP 工具调用结果封装为 ToolResponse 对象，取决于 wrap_tool_result 设置
        if self.wrap_tool_result:
            return ToolResponse(
                content=convert_to_as_blocks(res.content),
                metadata=res.meta
            )
        return res

而在 register_tool_function() 实现中，对于 MCPToolFunction 类型的可调用对象，直接获取其 json schema 即可：

def register_tool_function():
    if isinstance(tool_func, MCPToolFunction):
        # 设置工具函数本身的名称
        input_func_name = tool_func.name
        # 将 `__call__` 方法作为最终调用的函数对象
        original_func = tool_func.__call__
        # 直接获取 json schema
        json_schema = json_schema or tool_func.json_schema
        mcp_name = tool_func.mcp_name

通过 Toolkit 管理 MCP 工具

之前说过，AgentScope 通过 Tookit 管理所有工具，也包括 MCP 工具，为了将 MCP 工具注册到 Toolkit 中，可以调用其提供的 register_mcp_client：

async def register_mcp_client():
    for mcp_tool in await mcp_client.list_tools():
        # 根据 toolname 获取 MCPToolFunction 对象
        func_obj = await mcp_client.get_callable_function(
            func_name=mcp_tool.name,
            wrap_tool_result=True,
        )

        ......

        # 将该 MCPToolFunction 注册到 self._tools
        self.register_tool_function(
            tool_func=func_obj,
            group_name=group_name,
            preset_kwargs=preset_kwargs,
            postprocess_func=postprocess_func,
            namesake_strategy=namesake_strategy,
        )

而 remove_mcp_clients 方法则用于移除指定 MCP client 所提供的工具。

其他说明

MCP client 提供了 get_callable_function 方法，可以根据工具名称直接获取 MCPToolFunction 对象，这就使得我们可以在 Agent 代码中直接手动调用某个工具，不一定总是依靠 LLM 来触发工具调用：

func_obj = await stateless_client.get_callable_function()
res = await func_obj(arg1=value1, arg2=value2)

另外也可以通过这种方法来单独将某个 MCPToolFunction 对象注册到 Toolkit 中。

以上我们就整体介绍了 AgentScope 的 MCP 模块实现，以及 Toolkit 是如何管理 MCP 工具的。接下来我们再来学习 Toolkit 是如何实现 Skill 的。

Agent Skill

Skill（技能） 这个是由 Anthropic 公司率先提出并系统化定义的，目的是为 LLM 提供一种模块化的能力扩展机制。可以把 Skill 理解为给 AI 准备的 专家级工作交接手册 或 能力扩展包。它不只是简单的提示词，而是将特定领域的专业知识、标准操作流程（SOP）和可执行脚本封装成一个独立的能力单元。当 LLM 需要执行相关任务时，会自动判断并加载合适的 Skill。

• Skill（技能）：是一套流程和规范，像一个 方法论库，教会 AI 如何规划步骤、处理特定任务
• Tool（工具）：则提供原子化的能力/功能接口，可以执行某个具体的动作

通过 Skill，可以将高频复用的 复杂工作流 固化为一个 一键调用 的能力单元。一个 Skill 本身就是一个文件夹，其核心是 SKILL.md 文件。其中，SKILL.md 顶部的 YAML 元数据（YAML frontmatter）至关重要，它定义了 skill 的名称、描述等基本信息，LLM 最开始只会加载每个 Skill 的元数据内容，并依靠这些元数据内容，来判断是否时候需要使用该 Skill。当 LLM 确定为了完成某个任务，需要用到该 Skill 时，才会完整加载该 Skill 的指令。这样就实现了：

• 按需加载（Load on Demand）：不会让 skill 占用过多上下文窗口
• 意图驱动检索（Intent-Driven Retrieval）：元数据中的 description 字段承担了 检索索引 的角色，LLM 通过语义匹配快速定位正确的 Skill
• 资源分级（Tiered Resources） —— SKILL.md 内部还可以进一步引用外部文件，实现更深层的按需加载

接下来我们再来分析 AgentScope 如何实现对 Skill 的支持。

Toolkit 如何管理 Skill

在 Toolkit 中，通过 skill 属性来管理所有的技能，它是一个字典类型，key 是 skill 的名称，value 是一个 AgentSkill 对象。

class Toolkit(StateModule):
    def __init__(self):
        # 存储 AgentSkill 的字典，key 是 name
        self.skills: dict[str, AgentSkill] = {}

# 定义 Agent skill 类
class AgentSkill(TypedDict):
    """The agent skill typed dict class"""

    name: str
    """The name of the skill."""
    description: str
    """The description of the skill."""
    dir: str
    """The directory of the agent skill."""

Toolkit 类提供 register_agent_skill() 方法来注册技能，它只有一个参数，即指定该 skill 所在目录的路径。register_agent_skill() 的核心逻辑就是读取该目录下的 SKILL.md 文件，并从中提取 YAML 前置部分的元数据信息，得到该 skill 的名称和描述，然后将其封装为一个 AgentSkill 对象，并存储到 self.skills 中。

def register_agent_skill(self, skill_dir: str) -> None:
    import frontmatter

    path_skill_md = os.path.join(skill_dir, "SKILL.md")

    # 读取 YAML 文件 的 frontmatter 部份
    with open(path_skill_md, "r", encoding="utf-8") as f:
        post = frontmatter.load(f)

    name = post.get("name", None)
    description = post.get("description", None)

    self.skills[name] = AgentSkill(
        name=name,
        description=description,
        dir=skill_dir,
    )

同样提供了 remove_agent_skill() 方法来移除技能：

def remove_agent_skill(self, name: str) -> None:
    """根据名称移除已注册的智能体技能"""
    if name in self.skills:
        self.skills.pop(name)

LLM 如何感知 SKILL

在向 Agent 注册技能后，LLM 又是如何感知这些技能的存在呢？答案就是 提示词工程。Toolkit 的 system_prompt 属性提供的系统提示词中，包含了 skill 相关的提示词内容。而通过属性的方式来访问 sys_prompt 属性时，使得系统提示词是动态计算的，技能发生变更后可以立即生效。

class ReActAgent(ReActAgentBase):
    @property
    def sys_prompt(self) -> str:
        """The dynamic system prompt of the agent."""
        agent_skill_prompt = self.toolkit.get_agent_skill_prompt()
        if agent_skill_prompt:
            return self._sys_prompt + "\n\n" + agent_skill_prompt
        else:
            return self._sys_prompt

可以看到，智能体的系统提示词包含两部分构成：

• 用户创建 ReActAgent 时所提供的系统提示词 _sys_prompt
• Toolkit 中技能相关的提示词

get_agent_skill_prompt 用来获取技能相关的提示词：

def get_agent_skill_prompt(self) -> str | None:
    # 如果没有 skills，直接返回
    if len(self.skills) == 0:
        return None

    skill_descriptions = [
        self._agent_skill_instruction,
    ] + [
        # 这里列出所有的 skill 的描述
        self._agent_skill_template.format(
            name=_["name"],
            description=_["description"],
            dir=_["dir"],
        )
        for _ in self.skills.values()
    ]
    return "\n".join(skill_descriptions)

从代码实现可以看出，技能提示词也由两个部分构成：

• 技能的整体指令 _agent_skill_instruction
• 和所有技能的描述信息：每个描述信息都是通过 _agent_skill_template 渲染得到

这两个属性都可以定制的，创建 Toolkit 对象时可以通过参数设置这两个属性：

class Toolkit(StateModule):
    def __init__(
        self,
        agent_skill_instruction: str | None = None,
        agent_skill_template: str | None = None,
    ) -> None:
        ......
        self._agent_skill_instruction = (
            agent_skill_instruction or self._DEFAULT_AGENT_SKILL_INSTRUCTION
        )
        self._agent_skill_template = (
            agent_skill_template or self._DEFAULT_AGENT_SKILL_TEMPLATE
        )

如果没有设置的话，技能的整体指令 默认内容是：

"# Agent Skills\n"
"The agent skills are a collection of folds of instructions, scripts, "
"and resources that you can load dynamically to improve performance "
"on specialized tasks. Each agent skill has a `SKILL.md` file in its "
"folder that describes how to use the skill. If you want to use a "
"skill, you MUST read its `SKILL.md` file carefully."

而默认的技能描述渲染模版则是，这个模版其实就是包含 Skill 名称、描述和路径信息。

## {name}
{description}
Check "{dir}/SKILL.md" for how to use this skill

实际例子

我们来通过一个实际例子来看下注册 Skill 之后， ReActAgent 的系统提示词到底是怎么样的：

# -*- coding: utf-8 -*-
import os
from agentscope.tool import Toolkit
from agentscope.agent import ReActAgent
from agentscope.memory import InMemoryMemory
from agentscope.formatter import DashScopeChatFormatter
from agentscope.model import DashScopeChatModel

SKILLS_DIR = os.path.join(os.path.dirname(__file__), "skills")

toolkit = Toolkit()
toolkit.register_agent_skill(os.path.join(SKILLS_DIR, "python_helper"))
toolkit.register_agent_skill(os.path.join(SKILLS_DIR, "data_analyzer"))

agent = ReActAgent(
    name="Assistant",
    sys_prompt="你是一个智能助手。",
    model=DashScopeChatModel(
        model_name="qwen-plus",
        api_key=os.environ.get("DASHSCOPE_API_KEY", "mock-key"),
    ),
    formatter=DashScopeChatFormatter(),
    toolkit=toolkit,
    memory=InMemoryMemory(),
)

print(agent.sys_prompt)

这里例子中，我们注册了两个技能：python_helper 和 data_analyzer。来看下最后生成的系统提示词：

你是一个智能助手。

# Agent Skills
The agent skills are a collection of folds of instructions, scripts, and resources that you can load dynamically to improve performance on specialized tasks. Each agent skill has a `SKILL.md` file in its folder that describes how to use the skill. If you want to use a skill, you MUST read its `SKILL.md` file carefully.
## Python Helper
Python 代码编写助手，帮助编写高质量 Python 代码
Check "/root/code/private/python/open_source/agentscope/ai_demo/skills/python_helper/SKILL.md" for how to use this skill
## Data Analyzer
数据分析助手，支持数据清洗、统计分析和可视化
Check "/root/code/private/python/open_source/agentscope/ai_demo/skills/data_analyzer/SKILL.md" for how to use this skill

技能小结

以上就分析了 AgentScope 是如何支持 Skill 机制的，包括如何注册、移除提示词，以及如何通过系统提示词来让 LLM 知道什么是 Agent Skills 以及当前已有 skill 的描述信息。最后需要注意一点，为了让 ReActAgent 能够在运行过程中顺利加载某个 skill 的具体指令（SKILL.md 文件），需要给 Agent 配置文件读取工具：

toolkit.register_tool_function(view_text_file)  # 文件读取工具
# 或者
toolkit.register_tool_function(execute_shell_command)  # shell 命令执行工具

小结

这篇文章我们分析了 AgentScope 是如何支持 MCP 工具和 Agent Skills 这两种机制的。通过这篇文章以及上一篇文章，我们就整体分析了 AgentScope 的 工具系统 的实现原理。下一篇文章我们将开始介绍 AgentScope 是如何处理支持各种 LLM 的。