AI导哈

AI代理协调工具

轻量级AI代理协调工具，支持多LLM提供商、多代理协调和分布式设置。

  • by hasanjawad001
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 hasanjawad001/agentic-ai-mcp

使用说明

代理AI MCP

使用MCP工具的轻量级代理AI。支持多个LLM提供程序（Anthropic、OpenAI）、多代理编排和分布式设置，其中工具在一台机器上运行，代理在另一台计算机上运行。

安装

pip install agentic-ai-mcp

设置

在中设置API密钥 .env 文件（仅在客户端/代理计算机上需要）：

# For Anthropic (default)
ANTHROPIC_API_KEY=sk-ant-...

# For OpenAI
OPENAI_API_KEY=sk-...

快速开始

请参阅示例笔记本：

用法

服务器（公开工具）

在要承载工具的计算机上运行以下命令：

from agentic_ai_mcp import AgenticAIServer

def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def greet(name: str, times: int = 1) -> str:
    """Greet someone."""
    return ("Hello, " + name + "! ") * times

# Create server and register tools
server = AgenticAIServer(host="0.0.0.0", port=8888)
server.register_tool(add)
server.register_tool(greet)

print(f"Tools: {server.tools}")
print(f"URL: {server.mcp_url}")

# Start server in background
server.start()

# ... do other things ...

# Stop when done
server.stop()

客户端（运行代理）

在另一台计算机上运行此程序以连接到服务器并执行代理：

from agentic_ai_mcp import AgenticAIClient

# Connect to MCP server
client = AgenticAIClient(mcp_url="http://<server-ip>:8888/mcp")

# Simple agent workflow
result = await client.run("Calculate 2+1, then greet 'Alice' that many times.")
print(result)

# Planning-based workflow for complex tasks
result = await client.run_with_planning("Calculate ((0+2) + (1+1) + 1), then greet 'Bob' that many times.")
print(result)

多个MCP服务器

连接到分布在多个服务器上的工具：

from agentic_ai_mcp import AgenticAIClient

# Connect to multiple MCP servers
client = AgenticAIClient(
    mcp_urls=[
        "http://<server-1>:8888/mcp",  # math tools
        "http://<server-2>:9999/mcp",  # greeting tools
    ]
)

# The agent can use tools from all servers
result = await client.run("Calculate 2+3, then greet 'Alice' that many times.")
print(result)

多代理编排

使用 AgenticAIOrchestrator 协调多个代理共同完成一项任务。每个代理都可以有一个 role 和一个 tool_filter 使其行为专业化。

顺序流

特工们一个接一个地跑。每个代理都将前一个代理的输出视为上下文：

from agentic_ai_mcp import AgenticAIClient, AgenticAIOrchestrator

researcher = AgenticAIClient(
    mcp_url="http://<server-ip>:8888/mcp",
    role="researcher",
    tool_filter=["search"],  # only load the 'search' tool
)

writer = AgenticAIClient(
    mcp_url="http://<server-ip>:8888/mcp",
    role="writer",
    tool_filter=["write_file"],
)

orchestrator = AgenticAIOrchestrator(
    clients=[researcher, writer],
    flow_type="sequential",
)

result = await orchestrator.run("Research Python best practices and write a summary.")
print(result)

平行流

代理同时运行，其结果被合并：

orchestrator = AgenticAIOrchestrator(
    clients=[agent_a, agent_b],
    flow_type="parallel",
)

result = await orchestrator.run("Analyze this dataset from two perspectives.")
print(result)

与规划

编排器还支持每个代理上的规划工作流程：

result = await orchestrator.run_with_planning("Complex multi-step task...")

共享状态

当编排多个代理时，请使用 SharedState 在他们之间共享数据：

from agentic_ai_mcp import AgenticAIOrchestrator, SharedState

state = SharedState({"project": "demo"})

orchestrator = AgenticAIOrchestrator(
    clients=[agent_a, agent_b],
    shared_state=state,
)

result = await orchestrator.run("Do the task.")

# After execution, agent results are stored in shared state
print(state.get("agent_results"))
print(state.to_dict())

使用OpenAI

from agentic_ai_mcp import AgenticAIClient

# Use OpenAI instead of Anthropic
client = AgenticAIClient(
    mcp_url="http://<server-ip>:8888/mcp",
    provider="openai",
    model="gpt-4o-mini"
)

result = await client.run("Calculate -1+2")

直接传递API密钥

from agentic_ai_mcp import AgenticAIClient

# Pass API key directly (instead of using .env)
client = AgenticAIClient(
    mcp_url="http://<server-ip>:8888/mcp",
    api_key="sk-ant-..."
)

同步使用

所有异步方法都有同步变体，可在异步上下文之外使用：

# Single client
result = client.run_sync("Calculate 2+3")
result = client.run_with_planning_sync("Complex task...")

# Orchestrator
result = orchestrator.run_sync("Do the task.")
result = orchestrator.run_with_planning_sync("Complex task...")

API 参考

代理服务器

属性/方法 描述
server.tools 已注册工具名称列表
server.mcp_url 服务器URL
server.is_running 检查服务器是否正在运行
server.register_tool(func) 将函数注册为MCP工具
server.start() 在后台启动MCP服务器
server.stop() 停止MCP服务器

代理客户

属性/方法 描述
client.mcp_url 主MCP服务器URL（列表中的第一个）
client.mcp_urls 所有MCP服务器URL列表
client.tools 从所有服务器加载的工具名称列表
client.role 代理的角色描述（由编排器使用）
client.shared_state 共享状态字典（由编排器设置， None 如果独立）
client.run(prompt) 简单代理工作流（异步）
client.run_with_planning(prompt) 基于规划的工作流（异步）
client.run_sync(prompt) 简单的代理工作流程（同步）
client.run_with_planning_sync(prompt) 基于计划的工作流（同步）

Agent AI编排器

属性/方法 描述
orchestrator.clients 编排好的客户端列表
orchestrator.shared_state SharedState实例
orchestrator.run(prompt) 在每个客户端上使用简单代理运行（异步）
orchestrator.run_with_planning(prompt) 在每个客户端上按计划运行（异步）
orchestrator.run_sync(prompt) 同步版本 run()
orchestrator.run_with_planning_sync(prompt) 同步版本 run_with_planning()

共享状态

属性/方法 描述
state.get(key, default) 从共享状态获取值
state.set(key, value) 在共享状态下设置值
state.to_dict() 返回完整状态的副本
key in state 检查是否存在密钥

关键能力

  • 多服务器 --单个客户端可以连接到分布在多个MCP服务器上的工具
  • 图像传输 --通过协议传输图像，在Jupyter中自动显示
  • 多客户端编排 --协调多个代理 sequential 或 parallel 带有可选功能的流 synthesis
  • 共享状态 --编排过程中代理之间的数据共享
  • 多种执行模式 --简单的代理流(run)对于简单的任务，多步骤规划(run_with_planning)对于复杂的

许可证

麻省理工学院

相关 MCP