项目简介

Agntcy应用SDK是一个多功能工厂中心，旨在帮助开发者构建和实例化多代理组件。它支持Agent-to-Agent (A2A) 和Model Context Protocol (MCP) 等代理协议，并抽象化了底层通信传输（如SLIM、NATS、StreamableHTTP），使得智能体通信更加灵活和可扩展。这个SDK不仅提供了MCP客户端，也提供了构建MCP服务器的能力。

主要功能点

协议互操作性: 支持A2A、MCP和FastMCP等多种代理协议，实现智能体之间的无缝通信。
多传输支持: 通过统一的抽象接口，支持SLIM、NATS、StreamableHTTP等多种消息传输机制，提高了通信的灵活性和可插拔性。
代理服务器构建: 允许开发者通过简单的API快速创建和运行MCP服务器，该服务器能够托管和管理LLM所需的工具、数据资源和Prompt模板。
代理客户端: 提供便捷的客户端接口，用于与MCP服务器进行交互，例如调用服务器上注册的工具、读取上下文资源或获取Prompt。
可观测性集成: 内置集成可观测性SDK，支持应用程序的分布式追踪，便于监控和调试智能体系统的行为。

安装步骤

环境准备: 确保您的系统已安装Python 3.8+ 和 'pip' 包管理器。

SDK安装: 通过'pip'安装Agntcy应用SDK：

pip install agntcy-app-sdk

（可选）从源代码安装: 如果您需要从源代码进行开发或测试，请按以下步骤操作：

git clone https://github.com/agntcy/app-sdk.git
pip install -e app-sdk

消息总线服务: 运行MCP服务器通常需要一个底层消息总线服务（如NATS或SLIM）。您可以使用Docker Compose启动预配置的服务（具体'docker-compose.yaml'文件请参考仓库的'infra/docker/'目录）：
```
docker-compose -f infra/docker/docker-compose.yaml up
```

服务器配置

MCP服务器是供MCP客户端连接使用的后端服务。MCP客户端连接MCP服务器时，需要知道服务器的启动方式。以下是一个MCP服务器的JSON配置示例，它描述了如何启动一个基于Agntcy SDK的FastMCP服务器实例：

{
  "server_name": "MyFastMCPServer",
  "command": "uv",
  "args": [
    "run",
    "python",
    "tests/server/fast_mcp_server.py",
    "--transport", "NATS",
    "--endpoint", "localhost:4222",
    "--name", "default/default/fastmcp",
    "--non-blocking"
  ],
  "description": "一个基于Agntcy SDK构建的FastMCP服务器实例，通过NATS传输协议在localhost:4222上提供MCP服务。它支持调用自定义工具和访问资源。"
}

'server_name': 服务器的友好名称，例如 "我的FastMCP服务器"。
'command': 启动服务器所需的命令行工具。在此示例中，使用'uv'工具来运行Python脚本。
'args': 传递给'command'的参数列表，用于配置服务器的具体行为和连接信息。
- '--transport': 指定服务器将使用的消息传输协议，例如'NATS'（高性能消息队列）或'SLIM'（安全低延迟交互消息）。
- '--endpoint': 传输协议的连接地址。例如，NATS的默认地址是'localhost:4222'，SLIM通常是一个HTTP地址如'http://localhost:46357'。
- '--name': 服务器在代理网络中的唯一标识符，采用“组织/命名空间/本地名称”的格式，例如'default/default/fastmcp'。
- '--non-blocking': 一个可选参数，指示服务器以非阻塞模式运行。
'description': 对服务器功能的简要描述。

基本使用方法

服务器端示例

以下代码展示了如何使用Agntcy SDK构建一个简单的FastMCP服务器，它注册了一个'get_forecast'工具，并通过NATS传输协议提供服务：

import asyncio
from agntcy_app_sdk.factory import AgntcyFactory, TransportTypes
from mcp.server.fastmcp import FastMCP # 从MCP库导入FastMCP服务器实现

async def start_my_mcp_server():
    # 初始化Agntcy工厂，用于创建各种组件
    factory = AgntcyFactory(enable_tracing=False)
    
    # 创建一个FastMCP服务器实例
    mcp_server_instance = FastMCP()

    # 使用装饰器注册一个名为'get_forecast'的工具
    @mcp_server_instance.tool()
    async def get_forecast(location: str) -> str:
        """获取指定地点的天气预报。"""
        # 这是一个模拟的天气预报工具
        return f"获取{location}的天气：温度30°C，湿度50%，晴朗。"

    # 创建一个NATS传输协议实例，指定NATS服务器地址和本服务器名称
    transport = factory.create_transport(
        transport_type=TransportTypes.NATS.value,
        endpoint="localhost:4222", # NATS服务器地址
        name="default/default/my_mcp_agent" # 本MCP服务器在网络中的名称
    )

    # 创建消息桥接，连接FastMCP服务器和NATS传输协议，并指定通信主题
    bridge = factory.create_bridge(mcp_server_instance, transport=transport, topic="my_mcp_topic")
    print("MCP服务器正在启动，等待客户端连接...")
    
    # 启动消息桥接，让服务器开始监听和处理请求
    await bridge.start(blocking=True) # blocking=True 会让服务器持续运行

if __name__ == "__main__":
    asyncio.run(start_my_mcp_server())

运行上述脚本后，您的MCP服务器将在NATS 'localhost:4222'上启动，并监听'my_mcp_topic'主题的请求。

客户端示例

以下代码展示了如何使用Agntcy SDK客户端连接到上述MCP服务器，并调用其注册的'get_forecast'工具：

import asyncio
from agntcy_app_sdk.factory import AgntcyFactory, TransportTypes

async def interact_with_mcp_server():
    # 初始化Agntcy工厂
    factory = AgntcyFactory()
    
    # 创建NATS传输协议实例，指定NATS服务器地址和客户端名称
    transport_instance = factory.create_transport(
        transport="NATS",
        endpoint="localhost:4222", # NATS服务器地址
        name="default/default/mcp_client" # 客户端在网络中的名称
    )

    # 创建FastMCP客户端，指定要连接的服务器主题和传输协议
    mcp_client = await factory.create_client(
        "FastMCP",
        agent_topic="my_mcp_topic", # 对应服务器端设置的通信主题
        transport=transport_instance,
        agent_url="http://localhost:8081/mcp" # FastMCP协议可能需要的HTTP URL（通常由FastMCP服务器内部Uvicorn服务提供）
    )

    async with mcp_client as client:
        print("连接MCP服务器成功，正在列出可用工具...")
        tools = await client.list_tools()
        print("可用工具:", tools)

        # 检查是否有名为'get_forecast'的工具
        if "get_forecast" in [t.get("name") for t in tools]:
            print("发现'get_forecast'工具，正在调用以获取上海的天气预报...")
            result = await client.call_tool("get_forecast", {"location": "上海"})
            print("工具调用结果:", result.get("structuredContent", {}).get("result"))
        else:
            print("未找到'get_forecast'工具。")
            
    # 关闭传输连接
    await transport_instance.close()

if __name__ == "__main__":
    asyncio.run(interact_with_mcp_server())

运行上述客户端代码，它将连接到之前启动的MCP服务器，列出所有可用工具，并调用'get_forecast'工具获取天气预报。

关键词