AI导哈

消息监听与编排服务

MCP Monorepo 包含消息监听通道和编排器两个服务，用于接收、标准化和转发外部消息，并管理会话状态和内存。

  • by balbonis
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 balbonis/listen_mcp_orchestrator

使用说明

MCP Monorepo–监听通道+MCP编排器

这是一个 单体仓库 包含两个与您的匹配的服务（块） 架构：

  • 第1块——收听频道

    • 接收外部消息（例如，来自web客户端）
    • 使它们正常化
    • 将它们转发给编排器
    • 是否 不 管理会话/状态/内存

  • 第2块——MCP编排器

    • 拥有 会话、状态和内存
    • 使用混凝土 SessionContext 内部结构
    • 将决定+回复返回给收听频道
    • 用于MCP集成（您可以插入 mcp 稍后打包）

高级数据流：

[User / External System]
        ↓
 Block 1: Listening Channel
        ↓ (normalized event)
 Block 2: MCP Orchestrator (Session + State + Memory)

此版本使用 一个回购, 一个虚拟，以及 没有Docker.

1.设置

python -m venv .venv
source .venv/bin/activate    # Windows: .venv\Scripts\activate
pip install -r requirements.txt

可选：复制环境文件

cp .env.example .env

2.运行MCP编排器（块2）

在一个终端中：

source .venv/bin/activate
make run-orchestrator
# or, manually:
# uvicorn apps.orchestrator.app.main:app --reload --port 7002

检查健康状况：

curl http://localhost:7002/health

3.运行收听频道（块1）

在另一个终端中：

source .venv/bin/activate
make run-listening
# or, manually:
# uvicorn apps.listening_channel.app.main:app --reload --port 7001

检查健康状况：

curl http://localhost:7001/health

4.端到端测试

呼叫 区块1；它转发给 区块2，管理 SessionContext 为了 user-123:web:

curl -X POST http://localhost:7001/events/incoming   -H "Content-Type: application/json"   -d '{
    "channel": "web",
    "user_id": "user-123",
    "text": "hello orchestrator"
  }'

预期形状：

{
  "status": "ok",
  "reply": {
    "decision": "reply",
    "reply_text": "Hello, user-123! (from MCP Orchestrator)",
    "memory_snapshot": {
      "session_id": "user-123:web",
      "turn_count": 1
    },
    "debug": {
      "version": "v1",
      "channel": "web"
    }
  }
}

5.项目结构

apps/
  listening_channel/         # Block 1
    app/
      __init__.py
      main.py                # FastAPI app (Listening Channel)
      config.py              # Settings
      models.py              # Pydantic schemas
  orchestrator/              # Block 2
    app/
      __init__.py
      main.py                # FastAPI app (MCP Orchestrator)
      config.py
      models.py              # Request/response for orchestrator API
      session_context.py     # SessionContext struct
      memory_store.py        # Simple in-memory store
      orchestration.py       # AgentCore using SessionContext
tests/
  test_listening_health.py
  test_orchestrator_health.py

6.测试

source .venv/bin/activate
pytest -q

您可以将此monorepo直接推送到GitHub，并对这两种服务进行改进 同时按文件夹保持干净的分隔。

相关 MCP