Hugging Face 发布 Python Tiny Agents:约 70 行代码实现 MCP 智能体
Tiny Agents in Python: a MCP-powered agent in ~70 lines of code
Hugging Face 发布 Python 版 Tiny Agents,通过扩展 huggingface_hub 的 MCPClient 让智能体从 MCP 服务器拉取工具并传给 LLM,核心代码约 70 行。
官方教程给出 70 行搭 MCP 智能体的完整配置与代码路径,可直接照此复现。
新功能:tiny-agents 现已支持 AGENTS.md 标准。🥳
受 Tiny Agents in JS 启发,我们将这一理念移植到了 Python 🐍,并扩展了 huggingface_hub 客户端 SDK,使其能作为 MCP Client 运行,从而可以从 MCP 服务器拉取工具,并在推理时传递给 LLM。
MCP(Model Context Protocol)是一个开放协议,用于标准化大型语言模型(LLM)与外部工具和 API 的交互方式。本质上,它省去了为每个工具编写自定义集成的需要,让为 LLM 接入新能力变得更简单。
在这篇博客文章中,我们将展示如何开始使用一个连接到 MCP 服务器的 Python 微型 Agent,以解锁强大的工具能力。你会看到,启动自己的 Agent 并开始构建是多么容易!
剧透:Agent 本质上就是一个直接构建在 MCP Client 之上的 while 循环!
如何运行演示
本节将引导你如何使用现有的 Tiny Agents。我们将介绍设置过程和让 agent 运行的命令。
首先,你需要安装最新版本的 huggingface_hub,并带有 mcp extra,以获得所有必要组件。
pip install "huggingface_hub[mcp]>=0.32.0"
现在,让我们使用 CLI 运行一个 agent!
最酷的部分是,你可以直接从 HF Mirror Hub 的 tiny-agents 数据集加载 agents,或者指定你自己本地 agent 配置的路径!
> tiny-agents run --help
Usage: tiny-agents run [OPTIONS] [PATH] COMMAND [ARGS]...
Run the Agent in the CLI
╭─ Arguments ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ path [PATH] Path to a local folder containing an agent.json file or a built-in agent stored in the 'tiny-agents/tiny-agents' HF Mirror dataset │
│ (https://hf-mirror.com/datasets/tiny-agents/tiny-agents) │
╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
╭─ Options ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
│ --help Show this message and exit. │
╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
如果你不提供特定 agent 配置的路径,我们的 Tiny Agent 将默认连接到以下两个 MCP 服务器:
- “经典”的 file system server,它可以访问你的 Desktop,
- 以及 Playwright MCP 服务器,它知道如何为你使用沙盒化的 Chromium 浏览器。
下面的示例展示了一个网页浏览 agent,它配置为通过 Nebius 推理提供商使用 Qwen/Qwen2.5-72B-Instruct 模型,并且配备了 playwright MCP 服务器,这使它能够使用网页浏览器!该 agent 配置通过指定 它在 tiny-agents/tiny-agents 中的路径 从 HF Mirror 数据集加载。
当你运行这个 agent 时,你会看到它加载,并列出它从已连接的 MCP 服务器中发现的工具。然后,它就准备好接受你的提示了!
本演示中使用的提示:
在 Brave Search 上为 HF inference providers 做一次网页搜索,打开第一个结果,然后给我一份 HF Mirror 支持的 inference providers 列表
你也可以将 Gradio Spaces 用作 MCP 服务器!下面的示例通过 Nebius 推理提供商使用 Qwen/Qwen2.5-72B-Instruct 模型,并连接到一个 FLUX.1 [schnell] 图像生成 HF Space 作为 MCP 服务器。该 agent 从 HF Mirror Hub 上 tiny-agents/tiny-agents 数据集中的配置加载。
本演示中使用的提示:
生成一张 1024x1024 的图像,内容是一只小小的宇航员在月球表面从蛋中孵化出来。
现在你已经了解了如何运行现有的 Tiny Agents,接下来的章节将深入介绍它们如何工作,以及如何构建你自己的 Agent。
Agent 配置
每个 agent 的行为(其默认模型、推理提供商、要连接到哪些 MCP 服务器,以及其初始系统提示)都由一个 agent.json 文件定义。你也可以在同一目录中提供一个自定义的 PROMPT.md,以获得更详细的系统提示。下面是一个示例:
agent.json
model 和 provider 字段指定了智能体所使用的 LLM 和推理提供商。
servers 数组定义了智能体将连接的 MCP 服务器。
在此示例中,配置了一个“stdio” MCP 服务器。这种类型的服务器作为本地进程运行。智能体使用指定的 command 和 args 启动它,然后通过 stdin/stdout 与其通信,以发现并执行可用的工具。
{
"model": "Qwen/Qwen2.5-72B-Instruct",
"provider": "nebius",
"servers": [
{
"type": "stdio",
"command": "npx",
"args": ["@playwright/mcp@latest"]
}
]
}
PROMPT.md
You are an agent - please keep going until the user’s query is completely resolved [...]
您可以在此处找到有关 HF Mirror Inference Providers 的更多详细信息。
LLM 可以使用工具
现代 LLM 专为函数调用(或工具使用)而构建,这使用户能够轻松构建针对特定用例和现实世界任务量身定制的应用程序。
函数由其 schema 定义,schema 告知 LLM 它的功能以及它期望的输入参数。LLM 决定何时使用工具,随后智能体负责协调运行工具并将结果反馈回去。
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current temperature for a given location.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City and country e.g. Paris, France"
}
},
"required": ["location"],
},
}
}
]
InferenceClient 实现了与 OpenAI Chat Completions API 相同的工具调用接口,后者是推理提供商和社区公认的标准。
构建我们的 Python MCP 客户端
MCPClient 是我们工具使用功能的核心。它现在是 huggingface_hub 的一部分,并使用 AsyncInferenceClient 与 LLM 进行通信。
如果您想跟随实际代码一起操作,完整的
MCPClient代码在这里 🤓
MCPClient 的主要职责:
- 管理到一个或多个 MCP 服务器的异步连接。
- 从这些服务器发现工具。
- 为 LLM 格式化这些工具。
- 通过正确的 MCP 服务器执行工具调用。
这里简要展示它如何连接到 MCP 服务器(add_mcp_server 方法):
# Lines 111-219 of `MCPClient.add_mcp_server`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L111:L219
class MCPClient:
...
async def add_mcp_server(self, type: ServerType, **params: Any):
# 'type' can be "stdio", "sse", or "http"
# 'params' are specific to the server type, e.g.:
# for "stdio": {"command": "my_tool_server_cmd", "args": ["--port", "1234"]}
# for "http": {"url": "http://my.tool.server/mcp"}
# 1. Establish connection based on type (stdio, sse, http)
# (Uses mcp.client.stdio_client, sse_client, or streamablehttp_client)
read, write = await self.exit_stack.enter_async_context(...)
# 2. Create an MCP ClientSession
session = await self.exit_stack.enter_async_context(
ClientSession(read_stream=read, write_stream=write, ...)
)
await session.initialize()
# 3. List tools from the server
response = await session.list_tools()
for tool in response.tools:
# Store session for this tool
self.sessions[tool.name] = session
# Add tool to the list of available tools and Format for LLM
self.available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.input_schema,
},
})
它支持用于本地工具(例如访问您的文件系统)的 stdio 服务器,以及用于远程工具的 http 服务器!它还兼容 sse,这是远程工具的先前标准。
使用工具:流式传输与处理
MCPClient 的 process_single_turn_with_tools 方法是 LLM 交互发生的地方。它通过 AsyncInferenceClient.chat.completions.create(..., stream=True) 将对话历史和可用工具发送给 LLM。
1. 准备工具并调用 LLM
首先,该方法确定 LLM 在当前轮次应知晓的所有工具——这包括来自 MCP 服务器的工具以及任何用于智能体控制的特殊“退出循环”工具;然后,它向 LLM 发起流式调用:
# Lines 241-251 of `MCPClient.process_single_turn_with_tools`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L241:L251
# Prepare tools list based on options
tools = self.available_tools
if exit_loop_tools is not None:
tools = [*exit_loop_tools, *self.available_tools]
# Create the streaming request to the LLM
response = await self.client.chat.completions.create(
messages=messages,
tools=tools,
tool_choice="auto", # LLM decides if it needs a tool
stream=True,
)
当来自 LLM 的块到达时,该方法会遍历它们。每个块都会立即被 yield,然后我们重建完整的文本响应和任何工具调用。
# Lines 258-290 of `MCPClient.process_single_turn_with_tools`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L258:L290
# Read from stream
async for chunk in response:
# Yield each chunk to caller
yield chunk
# Aggregate LLM's text response and parts of tool calls
…
2. 执行工具
一旦流完成,如果 LLM 请求了任何工具调用(现在已完全在 final_tool_calls 中重建),该方法会处理每一个:
# Lines 293-313 of `MCPClient.process_single_turn_with_tools`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/mcp_client.py#L293:L313
for tool_call in final_tool_calls.values():
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments or "{}")
# Prepare a message to store the tool's result
tool_message = {"role": "tool", "tool_call_id": tool_call.id, "content": "", "name": function_name}
# a. Is this a special "exit loop" tool?
if exit_loop_tools and function_name in [t.function.name for t in exit_loop_tools]:
# If so, yield a message and terminate this turn's processing
messages.append(ChatCompletionInputMessage.parse_obj_as_instance(tool_message))
yield ChatCompletionInputMessage.parse_obj_as_instance(tool_message)
return # The Agent's main loop will handle this signal
# b. It's a regular tool: find the MCP session and execute it
session = self.sessions.get(function_name) # self.sessions maps tool names to MCP connections
if session is not None:
result = await session.call_tool(function_name, function_args)
tool_message["content"] = format_result(result) # format_result processes tool output
else:
tool_message["content"] = f"Error: No session found for tool: {function_name}"
tool_message["content"] = error_msg
# Add tool result to history and yield it
...
它首先检查所调用的工具是否会退出循环(exit_loop_tool)。如果不是,它会找到负责该工具的正确 MCP 会话并调用 session.call_tool()。然后,结果(或错误响应)会被格式化,添加到对话历史中,并被 yield,以便智能体知晓工具的输出。
我们的小型 Python 智能体:它(几乎)只是一个循环!
由于 MCPClient 完成了所有工具交互的工作,我们的 Agent 类变得非常简单。它继承自 MCPClient 并添加了对话管理逻辑。
Agent 类很小,专注于对话循环,代码可以在这里找到。
1. 初始化智能体
当创建一个 Agent 时,它会接收一个 agent 配置(模型、提供商、要使用哪些 MCP 服务器、系统提示词),并用系统提示词初始化对话历史。然后,load_tools() 方法会遍历服务器配置(在 agent.json 中定义),并为每一个调用 add_mcp_server(来自父级 MCPClient),从而填充 agent 的工具箱。
# Lines 12-54 of `Agent`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/agent.py#L12:L54
class Agent(MCPClient):
def __init__(
self,
*,
model: str,
servers: Iterable[Dict], # Configuration for MCP servers
provider: Optional[PROVIDER_OR_POLICY_T] = None,
api_key: Optional[str] = None,
prompt: Optional[str] = None, # The system prompt
):
# Initialize the underlying MCPClient with model, provider, etc.
super().__init__(model=model, provider=provider, api_key=api_key)
# Store server configurations to be loaded
self._servers_cfg = list(servers)
# Start the conversation with a system message
self.messages: List[Union[Dict, ChatCompletionInputMessage]] = [
{"role": "system", "content": prompt or DEFAULT_SYSTEM_PROMPT}
]
async def load_tools(self) -> None:
# Connect to all configured MCP servers and register their tools
for cfg in self._servers_cfg:
await self.add_mcp_server(**cfg)
2. Agent 的核心:循环
Agent.run() 方法是一个异步生成器,用于处理单个用户输入。它管理对话轮次,决定 agent 当前任务何时完成。
# Lines 56-99 of `Agent.run()`
# https://github.com/huggingface/huggingface_hub/blob/main/src/huggingface_hub/inference/_mcp/agent.py#L56:L99
async def run(self, user_input: str, *, abort_event: Optional[asyncio.Event] = None, ...) -> AsyncGenerator[...]:
...
while True: # Main loop for processing the user_input
...
# Delegate to MCPClient to interact with LLM and tools for one step.
# This streams back LLM text, tool call info, and tool results.
async for item in self.process_single_turn_with_tools(
self.messages,
...
):
yield item
...
# Exit Conditions
# 1. Was an "exit" tool called?
if last.get("role") == "tool" and last.get("name") in {t.function.name for t in EXIT_LOOP_TOOLS}:
return
# 2. Max turns reached or LLM gave a final text answer?
if last.get("role") != "tool" and num_turns > MAX_NUM_TURNS:
return
if last.get("role") != "tool" and next_turn_should_call_tools:
return
next_turn_should_call_tools = (last_message.get("role") != "tool")
在 run() 循环内部:
- 它首先将用户提示词添加到对话中。
- 然后调用
MCPClient.process_single_turn_with_tools(...)以获取 LLM 的响应,并为一个推理步骤处理任何工具执行。 - 每个条目都会被立即 yield,从而实现对调用方的实时流式传输。
- 每一步之后,它会检查退出条件:是否使用了特殊的“exit loop”工具、是否达到最大轮次限制,或者 LLM 是否提供了似乎对当前请求而言是最终回复的文本响应。
后续步骤
有很多很酷的方式可以探索和扩展 MCP Client 和 Tiny Agent 🔥 这里有一些想法可以帮助你入门:
- 基准测试不同 LLM 模型和推理提供商如何影响 agent 性能:工具调用性能可能会有所不同,因为每个提供商可能以不同方式对其进行优化。你可以在这里找到受支持提供商的列表。
- 使用本地 LLM 推理服务器运行 tiny agents,例如 llama.cpp 或 LM Studio。
- ..当然还有贡献!在 HF Mirror Hub 上的 tiny-agents/tiny-agents 数据集中分享你独特的 tiny agents 并提交 PR。
欢迎提交拉取请求和贡献!再次强调,这里的一切都是开源的!💎❤️
来源:Hugging Face Blog · hf-mirror.com