MCP 服务器

MUXI Framework

一个可扩展的 AI 代理框架 (Yī gè kě kuòzhǎn de AI dàilǐ kuàngjià)

ranaroussi

开发者工具

访问服务器

README

MUXI 框架

MUXI 框架是一个多功能的 Python 框架，用于构建 AI 代理和多代理系统。

MUXI 框架是一个强大的平台，用于构建具有记忆、MCP 服务器集成和实时通信能力的 AI 代理。它通过统一的架构集成多个接口，为创建高级 AI 应用程序提供了坚实的基础。

[!WARNING] 本项目正在开发中，尚未准备好用于生产环境。我们正在积极开发该框架并添加新功能。请参阅路线图以获取有关项目当前状态及其发展方向的信息。

特性

🤖 多代理支持：创建和管理具有不同功能的多个 AI 代理
🧠 记忆系统：用于上下文交互的短期和长期记忆
- FAISS 用于短期缓冲记忆
- 带有 pgvector 的 PostgreSQL 用于可扩展的长期记忆
- 带有 sqlite-vec 的 SQLite 用于本地或轻量级部署
🔌 MCP 客户端集成：通过模型上下文协议服务器连接到外部服务
- 支持基于 Web 的 MCP 服务器的 HTTP+SSE 传输
- 支持本地可执行服务器的命令行传输
- 具有指数退避的强大重新连接机制
- 全面的错误处理和诊断
- 取消对长时间运行操作的支持
🌟 MCP 服务器实现：将您的代理公开为 MCP 兼容的服务器
- 基于 SSE 的 MCP 主机集成服务器端点
- 从代理功能中发现工具
- 与 REST API 共享的身份验证
- 非 SSE 客户端的桥接包
🔄 代理到代理 (A2A) 通信：启用代理之间的结构化通信
- 用于代理间通信的标准化消息格式
- 功能注册和发现
- 专门代理之间的任务委派
- 具有适当隔离的上下文共享
- 安全和身份验证
🌐 多个接口：REST API、WebSockets、CLI、Web UI 等。
🔄 智能消息路由：自动将消息定向到最合适的代理
📊 多用户支持：用于多租户应用程序的用户特定内存分区
📘 上下文记忆：存储和检索结构化信息以个性化响应
🔍 代理级别知识库：通过轻量级 RAG 为代理提供专门的领域知识
🔄 混合通信协议：HTTP 用于标准请求，SSE 用于流式传输，WebSockets 用于多模式
📝 声明式配置：使用 YAML 或 JSON 文件定义代理，只需最少的代码
🚀 模块化架构：仅使用您需要的组件
🌍 特定于语言的 SDK（将在 v0.5.0 中推出）：TypeScript/JavaScript、Go 等的客户端库

架构

MUXI 具有灵活的、面向服务的方法：

┌───────────────────┐
│      客户端      │
│ (CLI/MCP/Web/SDK) │
└─────────┬─────────┘
          │
          │  (API/SSE/WS)
          │
┌─────────│───────────────────────────────────────────┐
│         │    MUXI 服务器 (本地/远程)             │
│         │                                           │
│         │        ┌───────────────┐                  │   ┌──────────────────┐
│         └───────>│  协调器     │----------------------│ 缓冲/长期记忆 │
│                  └───────┬───────┘                  │   └──────────────────┘
│         ┌────────────────┼────────────────┐         │
│         │                │                │         │
│  ┌──────▼──────┐  ┌──────▼──────┐  ┌──────▼──────┐  │   ┌──────────────────┐
│  │   代理 1   │  │   代理 2   │  │   代理 N   │------│ 领域知识       │
│  └───┬─────↑───┘  └──────↑──────┘  └───↑─────┬───┘  │   └──────────────────┘
│      │     │             │             │     │      │
│      │     │             ↓             │     │      │
│      │     └─────────> (A2A) <─────────┘     │      │
│      │                                       │      │
│      │            ┌─────────────┐            │      │
│      └───────────>│ MCP 处理程序 │<───────────┘      │
│                   └──────┬──────┘                   │
└──────────────────────────│──────────────────────────┘
                           │
                           │ (gRPC/HTTP/Command)
                           │
┌──────────────────────────▼──────────────────────────┐
│                     MCP 服务器                     │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐  │
│  │   天气     │  │  网络搜索   │  │     ....    │  │
│  └─────────────┘  └─────────────┘  └─────────────┘  │
└─────────────────────────────────────────────────────┘

有关更多详细信息，请参阅架构文档。

安装

用于开发

# 克隆存储库
git clone https://github.com/yourusername/muxi-framework.git
cd muxi-framework

# 以开发模式安装
./install_dev.sh

从 PyPI (未来)

pip install muxi

快速开始

基于配置的方法

最简单的入门方法是基于配置的方法：

from muxi import muxi
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 使用协调器级别的内存配置初始化 MUXI
app = muxi(
    buffer_size=10,  # 缓冲区窗口大小为 10 条消息
    long_term="sqlite:///data/memory.db"  # 使用 SQLite 启用长期记忆
    # 或者：long_term="postgresql://user:pass@localhost/muxi"
)

# 从配置文件添加代理
app.add_agent("assistant", "configs/assistant.yaml")

# 与代理聊天
response = await app.chat("你好，你是谁？")
print(response)

# 运行服务器
# app.run()

示例配置文件 (configs/muxi_config.yaml):

# 顶层内存配置
memory:
  buffer_size: 10  # 缓冲区窗口大小为 10
  long_term: true  # 使用应用程序根目录中的默认 SQLite 启用长期记忆
  # 或指定 SQLite：long_term: "sqlite:///data/memory.db"
  # 或指定 PostgreSQL：long_term: "postgresql://user:pass@localhost/muxi"

# 代理定义为数组
agents:
  - name: assistant
    system_message: 你是一个乐于助人的 AI 助手。
    model:
      provider: openai
      api_key: "${OPENAI_API_KEY}"
      model: gpt-4o
      temperature: 0.7
    knowledge:
      - path: "knowledge/domain_facts.txt"
        description: "此代理的专门领域知识"
    mcp_servers:
    - name: web_search
      url: http://localhost:5001
      credentials:
      - id: search_api_key
        param_name: api_key
        required: true
        env_fallback: SEARCH_API_KEY

编程方法

您还可以使用 Orchestrator 接口以编程方式创建代理：

from muxi.core.orchestrator import Orchestrator
from muxi.models.providers.openai import OpenAIModel
from muxi.server.memory.buffer import BufferMemory
from muxi.server.memory.long_term import LongTermMemory
from muxi.knowledge.base import FileKnowledge

# 创建具有内存配置的协调器
buffer_memory = BufferMemory(max_messages=10)
long_term_memory = LongTermMemory(connection_string="postgresql://user:pass@localhost/muxi")
# 或使用 SQLite：LongTermMemory(connection_string="sqlite:///data/memory.db")

orchestrator = Orchestrator(
    buffer_memory=buffer_memory,
    long_term_memory=long_term_memory
)

# 创建一个基本代理（内存由协调器提供）
agent = orchestrator.create_agent(
    agent_id="assistant",
    model=OpenAIModel(model="gpt-4o", api_key="your_api_key"),
    system_message="你是一个乐于助人的 AI 助手。",
    description="用于回答问题和提供信息的多用途助手。"
)

# 向代理添加领域知识
product_knowledge = FileKnowledge(
    path="knowledge/products.txt",
    description="产品信息和规格"
)
await agent.add_knowledge(product_knowledge)

# 创建其他专门的代理
orchestrator.create_agent(
    agent_id="researcher",
    model=OpenAIModel(model="gpt-4o", api_key="your_api_key"),
    system_message="你是一个乐于助人的研究助手。",
    description="专门从事研究任务、数据分析和信息检索。"
)

orchestrator.create_agent(
    agent_id="local_assistant",
    model=OpenAIModel(model="gpt-4o", api_key="your_api_key"),
    system_message="你是一个乐于助人的私人助手。",
    description="用于任务、提醒和一般信息的私人助手。"
)

# 将 MCP 服务器添加到代理
await orchestrator.agents["assistant"].connect_mcp_server(
    name="web_search",
    url="http://localhost:5001",
    credentials={"api_key": "your_search_api_key"}
)

# 与代理聊天（基于查询自动路由）
response = await orchestrator.chat("告诉我关于量子物理学的事情")
print(response.content)

# 与特定代理聊天
response = await orchestrator.chat("告诉我关于量子物理学的事情", agent_name="researcher")
print(response.content)

# 支持多用户聊天
response = await orchestrator.chat("记住我的名字是 Alice", user_id="user123")
print(response.content)

# 运行服务器
# app.run()

使用 CLI

# 启动与默认代理的聊天会话
muxi chat

# 发送一次性消息
muxi send "法国的首都是什么？"

# 运行服务器
muxi run

使用服务器（带有 Websockets + MCP 服务器的 API）

启动服务器：

muxi run

然后，在另一个终端或从您的应用程序：

# 向代理发送消息
curl -X POST http://localhost:5050/agents/assistant/messages \
  -H "Content-Type: application/json" \
  -d '{"content": "法国的首都是什么？"}'

使用 Web UI

Web UI 需要先安装 web 模块：

# 安装 Web UI 模块
pip install muxi-web

# 启动具有 Web UI 支持的服务器
muxi run

然后打开您的浏览器并导航到：

http://localhost:5050

或者，如果您已经在其他地方运行了 MUXI 服务器，则可以单独运行 Web UI：

# 仅启动 Web UI，连接到服务器
muxi-web --server-url http://your-server-address:5050

核心功能

使用 MCP 服务器

该框架提供了一个强大的接口，用于使用模型上下文协议 (MCP) 服务器：

from muxi import muxi

# 使用内存配置初始化 MUXI
app = muxi(
    buffer_size=10,
    long_term=True  # 使用默认的 SQLite 数据库
)

# 添加具有 MCP 服务器功能的代理
app.add_agent("assistant", "configs/assistant.yaml")

# 连接到具有凭据的基于 HTTP 的 MCP 服务器
await app.get_agent("assistant").connect_mcp_server(
    name="weather",
    url="http://localhost:5001",  # 用于 HTTP+SSE 传输
    credentials={"api_key": "your_weather_api_key"}  # 可选：如果不需要身份验证，则可以省略
)

# 连接到没有凭据的基于命令行的 MCP 服务器
await app.get_agent("assistant").connect_mcp_server(
    name="calculator",
    command="npx -y @modelcontextprotocol/server-calculator",
    # 对于不需要身份验证的服务器，不需要凭据参数
)

# 使用 MCP 服务器功能与代理聊天
response = await app.chat(
    "纽约的天气怎么样？另外，123 * 456 是多少？",
    user_id="user123"
)
print(response.content)  # 代理将使用两个 MCP 服务器来回答

YAML 中的示例 MCP 服务器配置：

# 顶层内存配置
memory:
  buffer_size: 10
  long_term: true  # 使用默认的 SQLite 数据库

# 代理配置
agents:
  - name: assistant
    system_message: 你是一个乐于助人的 AI 助手。
    model:
      provider: openai
      model: gpt-4o
    mcp_servers:
    - name: web_search
      url: http://localhost:5001
      credentials:  # 可选：如果不需要身份验证，则可以省略
      - id: search_api_key
        param_name: api_key
        required: true
        env_fallback: SEARCH_API_KEY
    - name: calculator
      command: npx -y @modelcontextprotocol/server-calculator
      # 对于不需要身份验证的服务器，不需要凭据部分

MCP 服务器实现

MUXI 不仅可以连接到 MCP 服务器，还可以将您的代理公开为 MCP 兼容的服务器：

from muxi import muxi

# 使用内存配置初始化 MUXI
app = muxi(
    buffer_size=10,
    long_term="sqlite:///data/memory.db"
)

# 从配置添加代理
app.add_agent("assistant", "configs/assistant.yaml")

# 在不同的端口上启动 API 和 MCP 服务器
app.run_server(
    port=5050,
    mcp=True,
    mcp_port=5051
)

# 或者只启动 MCP 服务器
app.start_mcp(port=5051)

这允许其他 MCP 客户端（如 Claude、Cursor 或其他 AI 助手）连接到您的 MUXI 代理并使用其功能。

智能消息路由

根据用户消息的内容自动将其路由到最合适的代理：

from muxi import muxi

# 使用内存和多个专门的代理初始化您的应用程序
app = muxi(
    buffer_size=10,
    long_term="postgresql://user:pass@localhost/muxi"
)

# 添加多个专门的代理
await app.add_agent("weather", "configs/weather_agent.yaml")
await app.add_agent("finance", "configs/finance_agent.json")
await app.add_agent("travel", "configs/travel_agent.yaml")

# 消息将自动路由到最合适的代理
response = await app.chat("本周末东京的天气预报是什么？")  # 天气代理
response = await app.chat("我现在应该投资科技股吗？")  # 金融代理
response = await app.chat("巴塞罗那最好的景点是什么？")  # 旅行代理

协调器分析每条消息的内容，并根据其专业化和描述智能地将其路由到最合适的代理。这意味着您无需指定哪个代理应处理每个请求 - 系统会自动找出答案。

通过环境变量配置路由系统：

ROUTING_LLM=openai
ROUTING_LLM_MODEL=gpt-4o-mini
ROUTING_LLM_TEMPERATURE=0.0

代理到代理 (A2A) 通信

启用专门代理之间的协作：

from muxi import muxi

# 使用内存和多个专门的代理初始化 MUXI
app = muxi(
    buffer_size=10,
    long_term="sqlite:///data/memory.db"
)

# 添加多个专门的代理
await app.add_agent("researcher", "configs/researcher_agent.yaml")
await app.add_agent("writer", "configs/writer_agent.yaml")
await app.add_agent("fact_checker", "configs/fact_checker_agent.yaml")

# 启用 A2A 通信（默认启用，但可以配置）
app.configure_a2a(enabled=True, scope="internal")

# 处理需要多个代理的复杂查询
response = await app.chat(
    "研究 2025 年量子计算的进展并撰写一份总结报告",
    user_id="user123"
)

# 幕后：
# 1. 消息路由到研究员代理
# 2. 研究员代理收集信息
# 3. 研究员委托给作家代理创建摘要
# 4. 作家创建草稿并委托给事实检查员进行验证
# 5. 最终验证的响应返回给用户

A2A 通信允许代理无缝协作，从而创建比任何单个代理单独完成的更强大的工作流程。

使用内存和上下文记忆

from muxi import muxi

# 使用内存配置初始化 MUXI
app = muxi(
    buffer_size=10,
    long_term="postgresql://user:pass@localhost/muxi"
)

# 从配置文件添加多个代理
await app.add_agent("weather", "configs/weather_agent.yaml")
await app.add_agent("assistant", "configs/assistant.yaml")

# 为特定用户添加上下文记忆
user_id = "user123"
knowledge = {
    "name": "Alice",
    "location": {"city": "New York", "country": "USA"},
    "preferences": {"language": "English", "units": "metric"}
}
await app.add_user_context_memory(user_id, knowledge)

# 使用个性化上下文聊天
# 注意：无需指定 agent_id - 协调器将选择合适的代理
response = await app.chat(
    "我所在城市的天气怎么样？",
    user_id=user_id
)
print(response.content)  # 使用来自上下文记忆的 Alice 的位置数据

# 对于特定于代理的内存操作，您可以指定 agent_id
memory_results = await app.search_memory(
    "用户询问了关于天气的什么？",
    agent_id="weather",  # 当您需要定位特定代理的内存时指定
    user_id=user_id,
    limit=5
)
print("相关记忆：", memory_results)

向量数据库支持

MUXI 支持多个向量数据库后端用于长期记忆：

使用带有 `sqlite-vec` 的 SQLite

非常适合本地开发、小规模部署或边缘环境：

# 在您的环境变量中（.env 文件）
USE_LONG_TERM_MEMORY=sqlite:///data/memory.db
# 或者对于应用程序根目录中的默认 SQLite 数据库
USE_LONG_TERM_MEMORY=true

# 或者在您的配置文件中（YAML）
memory:
  buffer_size: 10
  long_term: "sqlite:///data/memory.db"
  # 或者对于应用程序根目录中的默认 SQLite 数据库
  # long_term: true

# 或者以编程方式
from muxi import muxi

app = muxi(long_term="sqlite:///data/memory.db")

使用带有 `pgvector` 的 PostgreSQL

建议用于生产部署或多用户环境：

# 在您的环境变量中（.env 文件）
POSTGRES_DATABASE_URL=postgresql://user:password@localhost:5432/muxi

# 或者在您的配置文件中（YAML）
memory:
  buffer_size: 10
  long_term: "postgresql://user:password@localhost:5432/muxi"

# 或者以编程方式
from muxi import muxi

app = muxi(long_term="postgresql://user:password@localhost:5432/muxi")

技术细节

通信协议

MUXI 实现了混合协议方法，以实现最佳性能和灵活性：

HTTP：用于标准 API 请求，如配置和管理
服务器发送事件 (SSE)：用于逐个令牌流式传输响应
WebSockets：用于具有双向通信的多模式功能

包结构

MUXI 框架被组织成一个模块化架构，包含以下组件：

muxi-framework/
├── packages/
│   ├── core/          # 核心组件：代理、内存、MCP 接口
│   ├── server/        # REST API 和 WebSocket 服务器
│   ├── cli/           # 命令行界面
│   ├── web/           # Web 用户界面
│   └── muxi/          # 集成所有组件的元包
└── tests/             # 所有组件的测试套件

特定于语言的 SDK（将在 v0.5.0 中推出）

MUXI 框架将为流行的编程语言提供客户端库：

TypeScript/JavaScript SDK：用于 Web 和 Node.js 应用程序
Go SDK：用于后端集成
其他计划的 SDK：Java/Kotlin、Rust、C#/.NET、PHP、Ruby

每个 SDK 将提供：

完整的 REST API 客户端实现
用于实时通信的 WebSocket 客户端
MCP 服务器协议实现
跨语言的一致的开发者体验

开发路线图

MUXI 框架的开发重点是以下优先级：

REST API 和 MCP 服务器实现 - 实现完整的 REST API，包括身份验证、流式传输和 API 文档
WebSocket API 实现 - 通过多模式消息支持增强实时通信
CLI 接口 - 改进命令行界面，支持所有 API 操作
Web UI - 开发具有实时更新的响应式 Web 界面
代理到代理通信 - 实现用于代理间通信的 A2A 协议
向量数据库增强 - 优化向量操作并支持其他数据库
LLM 提供商 - 扩展对各种 LLM 提供商的支持（Anthropic、Gemini、Grok、本地模型）
测试和文档 - 所有组件的全面测试和文档
部署和包分发 - Docker 容器化、Kubernetes 部署和 CI/CD 管道
特定于语言的 SDK - 开发 TypeScript、Go、Java/Kotlin、Rust 和 C#/.NET 的客户端库
多模式功能 - 添加对文档、图像和音频处理的支持
安全增强 - 为企业级部署实施高级安全功能

路线图文件提供了有关路线图的更多详细信息。

许可证

本项目采用双重许可模式，以平衡开源协作与可持续商业实践。

开发阶段（1.0 版本之前）

在开发阶段，该软件根据 Creative Commons Attribution-NonCommercial-NoDerivatives 4.0 (CC BY-NC-ND 4.0) 许可。此许可证禁止商业用途、衍生作品和再分发，以确保开发过程的完整性，并避免项目在成熟之前出现碎片化。

1.0 版本发布后

当项目达到 1.0 版本时，它将切换到更宽松的开源许可证，该许可证允许非商业用例的无限制使用和商业用例的广泛使用。

贡献

欢迎贡献！ 请阅读我们的贡献指南以了解有关我们的行为准则、开发设置以及提交拉取请求的流程的详细信息。

致谢

感谢使该框架成为可能的许多开源项目 ❤️