MCP¶
MCP 让 Agent 使用外部服务提供的工具。你不需要在应用里重写工具函数，只需连接服务并选择要开放的工具。
启动一个示例服务¶
为了直接体验，仓库提供一个本地课程查询服务。在第一个终端运行：
python examples/tools/mcp_server.py
from fastmcp import FastMCP
mcp = FastMCP("course-demo")
@mcp.tool()
def get_course_info(course: str) -> str:
"""查询课程的时间和地点。可查询的课程名：Python 入门。"""
if course == "Python 入门":
return "周一 10:00，教学楼 A101"
return "没有找到这门课程"
if __name__ == "__main__":
mcp.run(transport="http", host="127.0.0.1", port=8000)
服务监听 http://127.0.0.1:8000/mcp。保持这个终端运行，在第二个终端配置模型环境变量。
让 Agent 调用远端工具¶
python examples/runtime/mcp.py
import asyncio
import os
import bumblehive
from bumblehive.protocols import MCPServerConfig
async def main() -> None:
config = bumblehive.RuntimeArguments(
model=os.environ["BUMBLEHIVE_MODEL"],
api_key=os.environ["BUMBLEHIVE_API_KEY"],
base_url=os.environ["BUMBLEHIVE_BASE_URL"],
skill_names=[],
mcp_servers=(MCPServerConfig(
name="courses",
url="http://127.0.0.1:8000/mcp",
enabled_tools=["get_course_info"],
),),
tool_names=["mcp_courses_get_course_info"],
)
async with bumblehive.from_config(config) as runtime:
result = await runtime.run("请查询课程“Python 入门”的上课时间和地点。")
print(result.error.message if result.error else result.final_content)
if __name__ == "__main__":
asyncio.run(main())
模型应通过 MCP 查到“周一 10:00，教学楼 A101”。这里有两个名称：
enabled_tools=["get_course_info"]选择从服务注册的工具。tool_names=["mcp_courses_get_course_info"]选择交给 Agent 的工具。名称由mcp_服务名_工具名组成。
换成自己的服务¶
修改 url、name 和工具名称。需要 HTTP 鉴权时，在 MCPServerConfig 中添加：
headers={"Authorization": f"Bearer {os.environ['MCP_TOKEN']}"}
enabled_tools=["*"] 可以注册服务的全部工具。连接在进入 Runtime 时建立，退出时关闭。
tool_timeout 控制 MCP 调用超时，默认 30 秒；它不是后台工具的“超时后继续等待结果”。