mh-gateway — 编排网关

核心网关服务，依赖 minimal-harness SDK。负责场景加载、用户权限校验、事件流归集，协调前端与各 worker 服务的通信。

• 版本：0.1.1

• 端口：8005

• Swagger：http://localhost:8005/docs

开发者指南：docs/dev-guide.md（中文） · docs/dev-guide.agent.md（英文，面向 Coding Agent）

企业适配指导：docs/customer-adaptation-guide.md（中文） · docs/customer-adaptation-guide.agent.md（英文，面向 Coding Agent）

构建分发：docs/build-guide.md

在 mh 生态中的位置

适配层架构

orchestration 通过 LifespanHook 接口与外部系统解耦。所有适配器通过 create_app() 参数注入：

UserAuthProvider、PermissionChecker、MetadataManager Protocol 定义在 mh_gateway 内。OutboundAuthProvider、M2MAuthProvider、ConfigProvider Protocol 也定义在 mh_gateway 内。

create_app() 工厂函数（客户部署入口）

create_app() 是 mh-gateway 的唯一入口，所有适配器通过 LifespanHook 参数注入：

import asyncio
import logging
from contextlib import asynccontextmanager

from fastapi import FastAPI
from mh_gateway import (
    ConfigManager, ConfigSchema, create_app,
)
from my_adapters import (
    CorpUserAuthProvider, CorpPermissionChecker, CorpRegistry,
)

# 1. 解析配置（env → 可选配置中心 → 报错）
config_mgr = ConfigManager()
settings = asyncio.run(config_mgr.resolve(ConfigSchema, prefix="ORCH"))

# 2. 配置 root logger（可选，不配置则使用 SDK 内置默认日志）
root = logging.getLogger()
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
))
root.addHandler(handler)
root.setLevel(logging.DEBUG)

# 3. 定义 Adapter LifespanHook（应用启动时注入）
@asynccontextmanager
async def token_verifier(app: FastAPI):
    app.state.adapters.token_verifier = CorpUserAuthProvider()
    yield

@asynccontextmanager
async def permission_checker(app: FastAPI):
    app.state.adapters.permission_checker = CorpPermissionChecker()
    yield

@asynccontextmanager
async def management_provider(app: FastAPI):
    app.state.adapters.management_provider = CorpRegistry()
    yield

# 4. 注入你的企业适配器
app = create_app(
    settings=settings,
    token_verifier=token_verifier,
    permission_checker=permission_checker,
    management_provider=management_provider,
)

省略的适配器参数会使用内置默认实现，适合开发和演示。

部署后以 uvicorn my_app:app 启动。

AppState — 运行时可访问适配器

所有注入的适配器实例通过 request.app.state.adapters 访问：

from mh_gateway import AppState

adapters: AppState = request.app.state.adapters
identity = await adapters.token_verifier.verify(request)
perms = await adapters.permission_checker.get_permissions(user_id)
agents = await adapters.management_provider.list_agents()

API

用户面 API

管理面 API（需 MetadataManager + 对应资源管理权限：manage:scene:* / manage:agent:* / manage:tool:*）

M2M 端点

AI 生成端点

开发模式端点（仅 dev_mode=true）

文件型 Tool（ToolScriptStore + 脚本上传）

除了在 UI 中以表单方式逐字段创建 tool，平台还支持把 Python 文件上传后自动解析为 tool：

1. 用户在管理 UI 拖拽 .py 文件 → POST /api/v1/management/tools/upload
2. 服务端用 AST 解析脚本顶部变量，校验必填字段（TOOL_NAME、TOOL_DESCRIPTION、TOOL_DISPLAY_NAME_LOCALE、TOOL_DESCRIPTION_LOCALE、TOOL_PARAMETERS、async def execute()）和 shebang 解释器是否可执行
3. 通过后把脚本文件保存到 ToolScriptStore（每个 app 实现各自的存储位置），并写入 ToolCreate.script_path 创建 ExternalScriptToolBinding
4. 运行时由 ExternalToolWrapper 在子进程中执行 execute()，每次 yield 序列化为 JSON 推到前端作为 ToolProgress 事件，支持取消和异常传播

每个 .py 文件对应一个 tool，禁止在文件里定义 register() 函数。最简模板：

TOOL_NAME = "greeter"
TOOL_DESCRIPTION = "Greet a user."
TOOL_DISPLAY_NAME_LOCALE = {"zh": "问候", "en": "Greeter"}
TOOL_DESCRIPTION_LOCALE = {"zh": "问候用户", "en": "Greet a user."}
TOOL_PARAMETERS = {
    "type": "object",
    "properties": {"name": {"type": "string"}},
    "required": ["name"],
}

async def execute(name: str):
    yield {"result": f"Hello, {name}!"}

AuditMiddleware

每个 Agent 执行周期自动记录审计日志（包括 agent_start/end、llm_start/end、tool_start/end/error、token 用量）。 日志级别为 INFO，可通过 orchestration.audit logger 配置。

AccessLogMiddleware

每个 HTTP 请求自动输出一条结构化 JSON 访问日志，包含 method、path、status、duration_ms、trace_id、user_id 等字段。 日志级别为 INFO，可通过 orchestration.access logger 配置。

监控指标

当 metrics_enabled=true 时，服务会自动注册 MetricsCollector，在内存中采集如下指标并通过后台定时任务推送到日志：

指标通过 AuditMiddleware 的生命周期钩子自动采集。可通过 /api/v1/metrics 获取实时快照。

指标持久化（MetricsRepository）

管理面数据概览（/api/v1/management/metrics）基于 可选 的持久化指标仓库，采用与 MetricsCollector 相同的单例注入模式，不修改 GatewayAdapters：

from mh_gateway.metrics_repo import MetricsRepository, LLMCallRecord, set_metrics_repo

class MyMetricsRepository(MetricsRepository):
    async def record_llm_call(self, record: LLMCallRecord) -> None: ...
    async def record_tool_call(self, record: ToolCallRecord) -> None: ...
    async def query_summary(self, date_from=None, date_to=None) -> MetricsSummary: ...
    async def close(self) -> None: ...

# 部署方在 lifespan hook 中注册（create_app 已支持 lifespan_hooks 参数）
set_metrics_repo(MyMetricsRepository())

• 记录写入：由独立的 MetricsPersistenceMiddleware（单一职责，区别于审计）在 on_llm_end / on_tool_end 钩子中写入，每条 LLM 调用记录一次（含 user/session/agent/scenario/provider/model/token/耗时）。
• 查询聚合：query_summary 按日期区间（YYYY-MM-DD）聚合调用次数、Token、Top N、模型性能。
• 存储介质：协议对后端完全中立 —— 单实例可用 SQLite/JSONL 文件，水平扩展时部署方可换共享数据库实现。
• 权限：端点需要 manage:metrics:* 权限（require_permission）。
• 未注册仓库时端点返回 503；中间件在未注册时完全静默。

PermissionMiddleware

每个 Agent 运行时自动校验工具调用权限。可通过 check(user_id, perm) 返回 bool 实现自定义逻辑。

环境变量

所有环境变量以 ORCH_ 为前缀：

LLM Provider 配置

LLM 配置通过 ORCH_PROVIDER_{NAME}__{KEY} 环境变量设置，不再使用旧的 ORCH_LLM_* 变量：

# 配置 OpenAI
export ORCH_PROVIDER_OPENAI__API_KEY=sk-xxx
export ORCH_PROVIDER_OPENAI__BASE_URL=https://api.openai.com/v1

# 配置 Anthropic
export ORCH_PROVIDER_ANTHROPIC__API_KEY=sk-ant-xxx

内置 provider：openai、anthropic、openai_viz（openai 的克隆）。

Agent 元数据的 provider 和 model 字段控制 per-agent 的 provider 选择。默认使用 openai。

外部配置对接

当客户有自己的配置中心（Apollo/Nacos/Consul）和密钥系统（HashiCorp Vault/AWS Secrets Manager）时，可通过 ConfigProvider 协议对接。

解析优先级

配置值按以下优先级解析（高 → 低）：

1. 环境变量（ORCH_*）— 最高优先级，运维可临时覆盖
2. 外接配置（ConfigProvider 实例，敏感与非敏感通过不同实例区分）— 来自配置中心
3. 代码默认值 — 若以上均未设置，使用 ConfigSchema 中的默认值

远程 key 重映射

ConfigManager.resolve() 支持 key_mapping 参数，将内部字段名重映射为客户配置中心的 key：

cfg = await config_mgr.resolve(
    MyConfig,
    prefix="my.registry",
    key_mapping={
        "db_path": "woa.orchestration.db.path",
    },
    sensitive_fields={"api_key"},
)

实现自定义 UserAuthProvider（对接企业 SSO）

verify() 收到的是完整的 FastAPI Request，可读 Cookie/Header/调外部 API：

from typing import Any
from mh_gateway.auth import UserAuthProvider, UserIdentity

class CorpSSOVerifier(UserAuthProvider):
    async def verify(self, request: Any) -> UserIdentity | None:
        # 1. 从 Cookie 中提取会话标识
        session_id = request.cookies.get("sessionid")
        if not session_id:
            return None
        # 2. 调用企业认证 API（request 还可读其他 header/query）
        user_info = await self._call_auth_api(session_id)
        if not user_info:
            return None
        # 3. 返回标准身份（extra_data 保留完整信息）
        return UserIdentity(
            user_id=user_info["employee_id"],
            username=user_info["name"],
            roles=user_info.get("roles", []),
            extra_data=user_info,
        )

HTTP Bearer token 由内置 _DefaultAuthProvider 从 request.headers["authorization"] 提取，客户使用 Cookie 时直接在 verify() 中读取 request.cookies 即可。

实现其他 Provider

from mh_gateway import ConfigProvider

class ApolloConfigProvider(ConfigProvider):
    async def get(self, key: str) -> str | None:
        return await apollo_client.get_value(key)

class VaultSecretResolver(ConfigProvider):
    async def get(self, key: str) -> str | None:
        return await vault_client.read_secret(key)

内置 Agent 样例 & 开发模式

设置 ORCH_DEV_MODE=true 后，服务会暴露 3 个内置样例 agent 以及开发调试用工具端点：

内置 Tool 包括 calculator、handoff、discover_agents、show_ui_meta、general_visualization、stop_agent。

triage agent 在进程内本地执行（无 endpoint_url），code-reviewer 和 writer 通过 M2M 端点执行。内置 agent 的 system_prompt 支持中英文，根据前端传来的 Accept-Language 自动适配。

生产环境请确保 ORCH_DEV_MODE 为 false（默认值），并通过 management_provider LifespanHook 注入企业自己的注册中心实现。

内置前端 UI（一站式部署）

设置 ORCH_DEV_MODE=true 后，FastAPI 会在 / 直接 serve 编译后的 SPA（单页应用），前提是 static/ 目录存在。

# 构建前端（SPA + 组件 bundle → 复制到 static/）
bash scripts/build-frontend.sh

# 启动（前端在 http://localhost:8005）
ORCH_DEV_MODE=true uv run uvicorn mh_gateway.main:app --port 8005

注意：前端静态文件需预先构建并放入 static/ 目录。ORCH_DEV_MODE=true 时服务会自动挂载前端并处理 SPA fallback 路由。

本地开发

# 带前端（先构建前端 SPA + 复制到 static/）
bash scripts/dev-standalone.sh

# 或仅后端（前端由 Vite 开发服务器提供热更新）
uv run uvicorn mh_gateway.main:app --port 8005
cd web-frontend && npm run dev

或使用项目根目录的 bash scripts/dev.sh 一键启动所有服务。

构建分发

cd packages/mh-gateway
uv build
# 产出 dist/mh_gateway-*.whl

客户 pip install 后，编写自己的启动文件注入适配器即可。

测试

uv run pytest packages/mh-gateway/tests -v