58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
"""TransportProtocol — MCP 传输层抽象接口。
|
|
|
|
定义所有传输层（streamable-http / stdio）共有的 list_tools / call_tool / 生命周期方法。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Any
|
|
|
|
from app.modules.tool_runtime.definition import ToolResult
|
|
|
|
|
|
class TransportProtocol(ABC):
|
|
"""MCP 传输层协议接口。"""
|
|
|
|
server_id: str
|
|
server_name: str
|
|
|
|
@abstractmethod
|
|
async def list_tools(self) -> list[dict[str, Any]]:
|
|
"""调用 tools/list，返回工具清单。"""
|
|
...
|
|
|
|
@abstractmethod
|
|
async def call_tool(self, name: str, arguments: dict[str, Any]) -> ToolResult:
|
|
"""调用 tools/call，返回 ToolResult。"""
|
|
...
|
|
|
|
async def start(self) -> None:
|
|
"""启动传输层（stdio 启子进程；HTTP no-op）。"""
|
|
return
|
|
|
|
async def stop(self) -> None:
|
|
"""停止传输层（stdio 杀子进程；HTTP no-op）。"""
|
|
return
|
|
|
|
async def restart(self) -> None:
|
|
"""重启传输层。"""
|
|
await self.stop()
|
|
await self.start()
|
|
|
|
@property
|
|
def is_running(self) -> bool:
|
|
"""传输层是否正在运行。"""
|
|
return True
|
|
|
|
async def get_status(self) -> dict[str, Any]:
|
|
"""返回当前状态信息。"""
|
|
return {
|
|
"server_id": self.server_id,
|
|
"name": self.server_name,
|
|
"transport": "unknown",
|
|
"status": "unknown",
|
|
"pid": None,
|
|
"uptime_seconds": None,
|
|
}
|