Agent架构设计：通过代码封装MCP调用的工程实践

1次阅读

没有评论

共计 1996 个字符，预计需要花费 5 分钟才能阅读完成。

最近在重构公司内部工具平台时，发现直接暴露 MCP（Microservice Control Plane）接口给业务系统会带来一系列棘手问题：

安全性风险：某个业务组误用管理员接口导致数据泄露，事后审计发现根本原因是接口权限控制颗粒度太粗
维护困难：当 MCP 接口升级时，需要同步修改所有调用方代码，某次变更导致 17 个服务同时报错
监控盲区：工具调用链路过长时，无法准确定位是 MCP 服务问题还是调用方参数错误

尝试过三种不同方案后，最终选择了 Agent 中间层设计：

直接调用 MCP（原始方案）
优点：开发简单，调用直接
缺点：前文提到的三大痛点全中
API Gateway 方案
优点：统一入口，自带限流鉴权
缺点：需要维护独立网关集群，对我们中小规模系统来说成本过高
Agent 中间层（最终方案）
优点：代码级封装，灵活度高
缺点：需要自行实现部分网关功能

以下是经过生产验证的 Agent 基础实现（关键代码已脱敏）：

class ToolAgent:
    _registry = {}  # 工具注册表

    @classmethod
    def tool(cls, name=None, permissions=None):
        """工具注册装饰器"""
        def decorator(func):
            tool_name = name or func.__name__
            cls._registry[tool_name] = {
                'func': func,
                'perms': permissions or []}
            return func
        return decorator

    @classmethod
    def execute(cls, tool_name, user, **kwargs):
        """执行入口"""
        # 权限校验
        if not cls._check_permissions(user, tool_name):
            raise PermissionError(f'User {user} cannot access {tool_name}')

        # 调用日志
        start_time = time.time()
        logger.info(f'TOOL_CALL_START|{tool_name}|{user}')

        try:
            result = cls._registry[tool_name]['func'](**kwargs)

            # 监控埋点
            metrics.timer(
                'tool_exec_time', 
                labels={'tool': tool_name}
            ).observe(time.time() - start_time)

            return result
        except Exception as e:
            logger.error(f'TOOL_CALL_FAIL|{tool_name}|{str(e)}')
            raise
        finally:
            logger.info(f'TOOL_CALL_END|{tool_name}|{user}')

关键设计点注释：

工具注册机制：通过装饰器模式实现声明式注册，保持代码整洁
RBAC 实现 ：_check_permissions 方法会验证用户角色是否匹配工具要求的权限
可观测性：同时记录日志和 Prometheus 指标

上线前我们做了充分验证：

在 8 核 16G 的测试环境：

直接调用 MCP 平均延迟：28ms
经过 Agent 层后平均延迟：31ms
99 分位延迟增加：≤5ms

集成 pybreaker 实现熔断器：

from pybreaker import CircuitBreaker

tool_breaker = CircuitBreaker(
    fail_max=5, 
    reset_timeout=60
)

@ToolAgent.tool(name='高危操作')
@tool_breaker
def critical_operation():
    # 业务代码

在 Grafana 配置的关键看板：