Agent架构设计：如何通过代码调用MCP实现工具定义的安全封装

1次阅读

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

最近在重构一个内部工具平台时，发现直接暴露 MCP（Microservice Control Platform）的工具定义会带来两个严重问题：

安全风险 ：前端直接调用 MCP 接口时，无法精细控制每个工具的操作权限，存在越权风险
维护困难 ：当 MCP 接口变更时，需要同步修改所有调用方的代码

# 反面示例：前端直接调用 MCP 接口
@app.route('/exec_tool')
def exec_tool():
    tool_name = request.args.get('tool')
    # 直接透传用户参数到 MCP
    return requests.post(f'http://mcp/api/{tool_name}', data=request.data)

我们引入 Agent 层作为中间件，其核心职责包括：

协议转换 ：将外部请求转换为 MCP 理解的格式
权限校验 ：基于 RBAC 模型控制工具访问权限
流量管控 ：实现限流和熔断机制
日志审计 ：记录所有操作流水

调用流程示意：

Client -> [Agent] -> [Auth] -> [Logger] -> [MCP Proxy] -> MCP Service

下面是带完整安全控制的 Agent 实现：

class MCPAgent:
    def __init__(self, mcp_endpoint):
        self.mcp = mcp_endpoint
        self.rate_limiter = RedisRateLimiter()

    def call_tool(self, user, tool_name, params):
        """
        :param user: UserContext 对象
        :param tool_name: 工具注册名称
        :param params: 参数字典
        :return: (status_code, response)
        """
        # 1. 权限校验
        if not user.has_permission(f'tool.{tool_name}.execute'):
            logging.warning(f'权限拒绝: {user.id} 尝试调用 {tool_name}')
            return 403, {'error': 'Forbidden'}

        # 2. 频率控制
        if not self.rate_limiter.check(user.id, tool_name):
            return 429, {'error': 'Too many requests'}

        # 3. 参数过滤
        safe_params = self._sanitize_params(tool_name, params)

        # 4. 调用 MCP
        try:
            resp = requests.post(f'{self.mcp}/v1/tools/{tool_name}',
                json=safe_params,
                headers={'X-Request-ID': generate_trace_id()},
                timeout=30
            )
            resp.raise_for_status()
            return resp.status_code, resp.json()
        except Exception as e:
            logging.error(f'MCP 调用失败: {str(e)}')
            return 502, {'error': 'Service unavailable'}

关键安全措施：

参数消毒 ：

# 防止 SQL 注入等攻击
def _sanitize_params(self, tool_name, raw_params):
    schema = self._get_tool_schema(tool_name)
    return SchemaValidator.validate(schema, raw_params)

访问控制 ：

# RBAC 权限检查示例
class UserContext:
    def has_permission(self, permission):
        return permission in self.role.permissions

根据场景选择调用方式：

同步调用 ：适用于需要即时响应的工具（如配置查询）
异步调用 ：适合耗时操作（超过 2 秒的任务），建议实现：

// Go 异步调用示例（使用 channel）func (a *Agent) AsyncCall(tool string, params map[string]interface{}) <-chan Result {resultChan := make(chan Result, 1)
    go func() {defer close(resultChan)
        // ... 调用逻辑...
        resultChan <- result
    }()
    return resultChan
}