MCP 服务器

AmritaCore 内嵌 Model Context Protocol 客户端（MCPClient / MultiClientManager），让任意 MCP 服务器成为工具来源。

服务器脚本格式

服务器脚本是 str | Path。远程服务器使用 extra+transport 语法 EXTRA+PROTOCOL://[user:pwd@]host[:port]/path,其中 EXTRA 选择传输类型, PROTOCOL 是 http 或 https:

格式	传输	说明
streamable+http(s)://host[:port]/path	Streamable HTTP	当前 MCP HTTP 标准(推荐)
sse+http(s)://host[:port]/path	HTTP + SSE(旧版)	user:pwd@ 变成 Basic Auth;单独的 user@ 变成 bearer token
sse://host[:port]/path	HTTP + SSE	sse+http://... 的简写
stdio://["cmd","arg1",...]	本地子进程	JSON 数组:第一个元素是命令,其余是参数
"path/to/server.py"	本地子进程	纯文件路径(stdio)
http(s)://...	自动检测	透传给 fastmcp,由它自动检测传输类型

示例:

"path/to/filesystem-server.py",               # stdio(文件路径)
'stdio://["uvx","mcp-server-git"]',          # stdio(命令)
"streamable+http://mcp.example.com/mcp",      # Streamable HTTP
"sse+http://mcp.example.com/sse",             # HTTP + SSE(旧版)
"sse://mcp.example.com/sse",                  # sse+http 简写
"sse+https://user:pwd@mcp.example.com/sse",   # 带 Basic Auth 的 SSE

注意: 传输关键字是 streamable 而不是 stream。像 stream+http://... 这样的 URL 不是可识别的 extra,会被透传给 fastmcp,而它无法解析——请改用 streamable+http(s)://。

标准用法：配置 + 加载

标准方式是配置驱动：在 FunctionConfig 里列出服务器，让 minimal_init() / load_amrita() 启动它们。

from amrita_core import minimal_init
from amrita_core.config import AmritaConfig, FunctionConfig

config = AmritaConfig(
    function_config=FunctionConfig(
        agent_mcp_client_enable=True,
        agent_mcp_server_scripts=[
            "path/to/filesystem-server.py",        # stdio
            "streamable+http://mcp.example.com/mcp",  # Streamable HTTP
        ],
    )
)
await minimal_init(config)  # 加载并初始化所有 MCP 客户端

之后每个 MCP 工具都像常规工具一样注册进全局工具管理器——agent 按名调用即可。

高级用法：直接驱动 ClientManager

ClientManager 是单例——load_amrita() 用的就是同一个实例。需要运行时 控制时自己驱动它：

from amrita_core.tools.mcp import ClientManager

manager = ClientManager()  # 单例

# 一次性：注册 + 立即连接。
await manager.initialize_this("path/to/server.py")

# 或批量（单个失败不影响其余）。
await manager.initialize_scripts_all(
    [
        "path/to/server-a.py",
        "sse+http://mcp.example.com/sse",
    ]
)

# 延迟：先注册，后连接（例如绑定到会话之后）。
manager.register_only(server_script="path/to/server-b.py")
await manager.initialize_all()

ClientManager 继承 MultiClientManager——后者把工具名映射到客户端、 重映射重复工具名，并暴露 unregister_client(script) / reinitialize_all() / update_tools(client)。会话隔离：自行创建 MultiClientManager 实例， 通过 ability 上下文挂到不同会话（见数据后端）。

直接使用 MCPClient

单个服务器可直接用 MCPClient——适合测试或一次性集成：

from amrita_core.tools.mcp import MCPClient, MultiClientManager

client = MCPClient("path/to/server.py", connection_ttl=120)  # 空闲 TTL 后关闭；-1 禁用

# 绑定到管理器（注册 + 加载工具）。
await client.bound_to(MultiClientManager())

# 或绕过 agent 直接调用工具。
result = await client.simple_call("list_files", {"path": "/tmp"})

connection_ttl 控制空闲关闭：ttl 秒未使用后连接关闭，下次调用自动重连。 -1 保持常开。 默认并发安全。 每个服务器保持一条所有工具共享的常驻连接。simple_call 在底层 fastmcp Client(带引用计数的可重入上下文管理器)上用 async with 包裹调用——因此并行调用(如 agent 的工具 runner)绝不会互相拆掉连接。连接 只在最后一个活跃调用退出后才被回收,再由 TTL 任务释放。

下一步

自定义 Tokenizer——接入自己的 tokenizer 用于用量统计。