hankin
423baff73b
- Docker bridge 网络隔离（8000 端口封死） - Gunicorn 4 Worker 多进程 - Alembic 数据库迁移基线 - 日志轮转 20m×3 - JWT 密钥 + DB 密码 + CORS 收紧 - 3-2-1 备份链路（NAS + R740-B 冷备） - 连接池 pool_pre_ping + pool_recycle=3600
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
"""
|
|
MCP 工具注册中心
|
|
提供 @register_tool 装饰器和全局 TOOL_REGISTRY
|
|
"""
|
|
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable, Coroutine
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.schemas.auth import CurrentUserPayload
|
|
|
|
|
|
@dataclass
|
|
class MCPToolMeta:
|
|
"""MCP 工具元信息"""
|
|
name: str
|
|
description: str
|
|
parameters: dict[str, Any] # JSON Schema 描述
|
|
handler: Callable[
|
|
[AsyncSession, CurrentUserPayload, dict[str, Any]],
|
|
Coroutine[Any, Any, "MCPToolResult"]
|
|
] | None = None
|
|
|
|
|
|
@dataclass
|
|
class MCPToolResult:
|
|
"""MCP 工具执行结果"""
|
|
success: bool = True
|
|
# 返回给 AI/前端的类型：text 或 action_card
|
|
response_type: str = "text" # "text" | "action_card"
|
|
data: dict[str, Any] = field(default_factory=dict)
|
|
message: str = ""
|
|
|
|
|
|
# ── 全局工具注册表 ────────────────────────────────────────
|
|
TOOL_REGISTRY: dict[str, MCPToolMeta] = {}
|
|
|
|
|
|
def register_tool(
|
|
name: str,
|
|
description: str,
|
|
parameters: dict[str, Any] | None = None,
|
|
):
|
|
"""装饰器：注册 MCP 工具到全局注册表"""
|
|
def decorator(fn: Callable):
|
|
meta = MCPToolMeta(
|
|
name=name,
|
|
description=description,
|
|
parameters=parameters or {},
|
|
handler=fn,
|
|
)
|
|
TOOL_REGISTRY[name] = meta
|
|
return fn
|
|
return decorator
|
|
|
|
|
|
def get_tools_manifest() -> list[dict[str, Any]]:
|
|
"""返回所有已注册工具的清单（供 Dify Agent 读取配置）"""
|
|
return [
|
|
{
|
|
"name": meta.name,
|
|
"description": meta.description,
|
|
"parameters": meta.parameters,
|
|
}
|
|
for meta in TOOL_REGISTRY.values()
|
|
]
|
|
|
|
|
|
async def execute_tool(
|
|
tool_name: str,
|
|
db: AsyncSession,
|
|
user: CurrentUserPayload,
|
|
params: dict[str, Any],
|
|
) -> MCPToolResult:
|
|
"""根据工具名称执行对应 handler"""
|
|
meta = TOOL_REGISTRY.get(tool_name)
|
|
if meta is None or meta.handler is None:
|
|
return MCPToolResult(
|
|
success=False,
|
|
message=f"工具 '{tool_name}' 未注册或无 handler",
|
|
)
|
|
return await meta.handler(db, user, params)
|