From 93a6672face4c3e07ee8d3cca9da603973f8da95 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Fri, 27 Mar 2026 00:15:32 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=88=A0=E9=99=A4=20=5Finternal=20?= =?UTF-8?q?Python=20=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除 protocols 子包 (abp, acp, lsp, mcp) - 删除 runtime 子包 (orchestrator, rust) - 删除 abc 基类 (abp, acp, lsp, mcp, gateway, orchestrator) - 删除 geteway 子包 - 删除 skills, stars, tools 子包 --- .../abc/abp/base_astrbot_abp_client.py | 57 -- .../abc/acp/base_astrbot_acp_client.py | 66 --- .../abc/acp/base_astrbot_acp_server.py | 68 --- astrbot/_internal/abc/base_astrbot_gateway.py | 73 --- .../abc/base_astrbot_orchestrator.py | 352 ------------- .../abc/lsp/base_astrbot_lsp_client.py | 114 ---- .../abc/mcp/base_astrbot_mcp_client.py | 95 ---- astrbot/_internal/geteway/__init__.py | 6 - astrbot/_internal/geteway/deps.py | 4 - astrbot/_internal/geteway/routes/inspector.py | 0 astrbot/_internal/geteway/routes/memory.py | 0 astrbot/_internal/geteway/routes/stats.py | 0 astrbot/_internal/geteway/server.py | 248 --------- astrbot/_internal/geteway/ws_manager.py | 103 ---- astrbot/_internal/protocols/abp/__init__.py | 5 - astrbot/_internal/protocols/abp/client.py | 93 ---- astrbot/_internal/protocols/acp/__init__.py | 6 - astrbot/_internal/protocols/acp/client.py | 220 -------- astrbot/_internal/protocols/acp/server.py | 223 -------- astrbot/_internal/protocols/lsp/__init__.py | 5 - astrbot/_internal/protocols/lsp/client.py | 263 ---------- astrbot/_internal/protocols/mcp/__init__.py | 63 --- astrbot/_internal/protocols/mcp/client.py | 486 ------------------ astrbot/_internal/protocols/mcp/config.py | 55 -- astrbot/_internal/protocols/mcp/tool.py | 45 -- astrbot/_internal/runtime/__init__.py | 3 - astrbot/_internal/runtime/__main__.py | 24 - astrbot/_internal/runtime/orchestrator.py | 164 ------ astrbot/_internal/runtime/rust/__init__.py | 18 - astrbot/_internal/runtime/rust/_core.pyi | 16 - astrbot/_internal/skills/__init__.py | 13 - astrbot/_internal/stars/__init__.py | 7 - .../_internal/stars/runtime_status_star.py | 127 ----- astrbot/_internal/tools/__init__.py | 5 - astrbot/_internal/tools/base.py | 332 ------------ astrbot/_internal/tools/builtin.py | 48 -- astrbot/_internal/tools/registry.py | 278 ---------- 37 files changed, 3685 deletions(-) delete mode 100644 astrbot/_internal/abc/abp/base_astrbot_abp_client.py delete mode 100644 astrbot/_internal/abc/acp/base_astrbot_acp_client.py delete mode 100644 astrbot/_internal/abc/acp/base_astrbot_acp_server.py delete mode 100644 astrbot/_internal/abc/base_astrbot_gateway.py delete mode 100644 astrbot/_internal/abc/base_astrbot_orchestrator.py delete mode 100644 astrbot/_internal/abc/lsp/base_astrbot_lsp_client.py delete mode 100644 astrbot/_internal/abc/mcp/base_astrbot_mcp_client.py delete mode 100644 astrbot/_internal/geteway/__init__.py delete mode 100644 astrbot/_internal/geteway/deps.py delete mode 100644 astrbot/_internal/geteway/routes/inspector.py delete mode 100644 astrbot/_internal/geteway/routes/memory.py delete mode 100644 astrbot/_internal/geteway/routes/stats.py delete mode 100644 astrbot/_internal/geteway/server.py delete mode 100644 astrbot/_internal/geteway/ws_manager.py delete mode 100644 astrbot/_internal/protocols/abp/__init__.py delete mode 100644 astrbot/_internal/protocols/abp/client.py delete mode 100644 astrbot/_internal/protocols/acp/__init__.py delete mode 100644 astrbot/_internal/protocols/acp/client.py delete mode 100644 astrbot/_internal/protocols/acp/server.py delete mode 100644 astrbot/_internal/protocols/lsp/__init__.py delete mode 100644 astrbot/_internal/protocols/lsp/client.py delete mode 100644 astrbot/_internal/protocols/mcp/__init__.py delete mode 100644 astrbot/_internal/protocols/mcp/client.py delete mode 100644 astrbot/_internal/protocols/mcp/config.py delete mode 100644 astrbot/_internal/protocols/mcp/tool.py delete mode 100644 astrbot/_internal/runtime/__init__.py delete mode 100644 astrbot/_internal/runtime/__main__.py delete mode 100644 astrbot/_internal/runtime/orchestrator.py delete mode 100644 astrbot/_internal/runtime/rust/__init__.py delete mode 100644 astrbot/_internal/runtime/rust/_core.pyi delete mode 100644 astrbot/_internal/skills/__init__.py delete mode 100644 astrbot/_internal/stars/__init__.py delete mode 100644 astrbot/_internal/stars/runtime_status_star.py delete mode 100644 astrbot/_internal/tools/__init__.py delete mode 100644 astrbot/_internal/tools/base.py delete mode 100644 astrbot/_internal/tools/builtin.py delete mode 100644 astrbot/_internal/tools/registry.py diff --git a/astrbot/_internal/abc/abp/base_astrbot_abp_client.py b/astrbot/_internal/abc/abp/base_astrbot_abp_client.py deleted file mode 100644 index 07397e983..000000000 --- a/astrbot/_internal/abc/abp/base_astrbot_abp_client.py +++ /dev/null @@ -1,57 +0,0 @@ -""" -ABP (AstrBot Protocol) client - in-process star communication. -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import Any - - -class BaseAstrbotAbpClient(ABC): - """ - ABP client: in-process star (plugin) communication. - - Stars register themselves; client delegates calls to registered instances. - - Subclass must implement: - - connect() -> None - - register_star(name, instance) -> None - - unregister_star(name) -> None - - call_star_tool(star, tool, args) -> Any - - shutdown() -> None - """ - - @property - @abstractmethod - def connected(self) -> bool: ... - - @abstractmethod - async def connect(self) -> None: - """Lightweight: just sets connected=True.""" - ... - - @abstractmethod - def register_star(self, star_name: str, star_instance: Any) -> None: - """Add star to internal registry.""" - ... - - @abstractmethod - def unregister_star(self, star_name: str) -> None: - """Remove star from registry (idempotent).""" - ... - - @abstractmethod - async def call_star_tool( - self, - star_name: str, - tool_name: str, - arguments: dict[str, Any], - ) -> Any: - """Delegate to star_instance.call_tool(tool_name, arguments).""" - ... - - @abstractmethod - async def shutdown(self) -> None: - """Set connected=False, cancel pending requests.""" - ... diff --git a/astrbot/_internal/abc/acp/base_astrbot_acp_client.py b/astrbot/_internal/abc/acp/base_astrbot_acp_client.py deleted file mode 100644 index 3085631e6..000000000 --- a/astrbot/_internal/abc/acp/base_astrbot_acp_client.py +++ /dev/null @@ -1,66 +0,0 @@ -""" -ACP (AstrBot Communication Protocol) client. - -Transport: TCP | Unix Socket -Messages: JSON with Content-Length header -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import Any - - -class BaseAstrbotAcpClient(ABC): - """ - ACP client: connects to ACP servers via TCP or Unix socket. - - Subclass must implement: - - connect() -> None - - connect_to_server(host, port) -> None - - connect_to_unix_socket(path) -> None - - call_tool(server, tool, args) -> Any - - send_notification(method, params) -> None - - shutdown() -> None - """ - - @property - @abstractmethod - def connected(self) -> bool: ... - - @abstractmethod - async def connect(self) -> None: ... - - @abstractmethod - async def connect_to_server(self, host: str, port: int) -> None: - """Connect via TCP.""" - ... - - @abstractmethod - async def connect_to_unix_socket(self, socket_path: str) -> None: - """Connect via Unix domain socket.""" - ... - - @abstractmethod - async def call_tool( - self, - server_name: str, - tool_name: str, - arguments: dict[str, Any], - ) -> Any: - """Call tool on server, return result.""" - ... - - @abstractmethod - async def send_notification( - self, - method: str, - params: dict[str, Any], - ) -> None: - """Send one-way notification.""" - ... - - @abstractmethod - async def shutdown(self) -> None: - """Close connection, cancel pending requests.""" - ... diff --git a/astrbot/_internal/abc/acp/base_astrbot_acp_server.py b/astrbot/_internal/abc/acp/base_astrbot_acp_server.py deleted file mode 100644 index 86ad51052..000000000 --- a/astrbot/_internal/abc/acp/base_astrbot_acp_server.py +++ /dev/null @@ -1,68 +0,0 @@ -""" -ACP (AstrBot Communication Protocol) server. - -Transport: TCP listening socket -Messages: JSON with Content-Length header -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from collections.abc import Callable -from typing import Any - - -class BaseAstrbotAcpServer(ABC): - """ - ACP server: listens for client connections, exposes tools. - - Subclass must implement: - - start(host, port) -> None - - register_tool(name, handler) -> None - - register_notification_handler(name, handler) -> None - - broadcast_notification(method, params) -> None - - shutdown() -> None - """ - - @property - @abstractmethod - def running(self) -> bool: - """True if server is accepting connections.""" - ... - - @abstractmethod - async def start(self, host: str = "127.0.0.1", port: int = 8765) -> None: - """Bind and listen. Block until shutdown.""" - ... - - @abstractmethod - def register_tool( - self, - name: str, - handler: Callable[..., Any], - ) -> None: - """Register async tool handler (receives params dict, returns result).""" - ... - - @abstractmethod - def register_notification_handler( - self, - name: str, - handler: Callable[..., Any], - ) -> None: - """Register async notification handler (receives params dict).""" - ... - - @abstractmethod - async def broadcast_notification( - self, - method: str, - params: dict[str, Any], - ) -> None: - """Send notification to all connected clients.""" - ... - - @abstractmethod - async def shutdown(self) -> None: - """Stop accepting, close all client connections.""" - ... diff --git a/astrbot/_internal/abc/base_astrbot_gateway.py b/astrbot/_internal/abc/base_astrbot_gateway.py deleted file mode 100644 index c67c498f9..000000000 --- a/astrbot/_internal/abc/base_astrbot_gateway.py +++ /dev/null @@ -1,73 +0,0 @@ -""" -AstrBot Gateway - HTTP/WebSocket API server. - -Built on FastAPI, provides: -- HTTP REST API (stats, inspector, config) -- WebSocket for real-time events -- Static file serving (dashboard) -- Authentication (JWT/API key) -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod - - -class BaseAstrbotGateway(ABC): - """ - Gateway: HTTP/WebSocket server built on FastAPI. - - ┌─────────────────────────────────────────────────────────┐ - │ FastAPI App │ - ├─────────────────────────────────────────────────────────┤ - │ REST Endpoints WebSocket │ - │ ├─ GET /api/stats ├─ /ws (connection manager)│ - │ ├─ GET /api/inspector/* │ │ - │ ├─ GET /api/memory/* │ │ - │ └─ ... │ │ - │ │ - │ Middleware: CORS, Auth, Logging │ - └─────────────────────────────────────────────────────────┘ - │ - ▼ - ┌─────────────────────────┐ - │ Orchestrator │ - │ (owns protocol clients)│ - └─────────────────────────┘ - - Routes (typical): - GET / → Dashboard static files - GET /api/stats → System statistics - GET /api/inspector/stars → List registered stars - WS /ws → WebSocket for real-time events - - serve() Lifecycle: - 1. Create FastAPI app - 2. Register routes - 3. Start WebSocket manager - 4. Bind to host:port - 5. Run ASGI server (uvicorn/hypercorn) - 6. Block until shutdown - 7. Close all connections - - Subclass must implement: - - serve(): start server, block until shutdown - """ - - @abstractmethod - async def serve(self) -> None: - """ - Start gateway server - blocks until shutdown. - - Should: - 1. Create FastAPI app with routes - 2. Configure CORS, auth middleware - 3. Start WebSocket connection manager - 4. Bind to ASTRBOT_PORT (default 6185) - 5. Run ASGI server - 6. Handle graceful shutdown on SIGTERM/SIGINT - - Raises: - OSError: address already in use - """ - ... diff --git a/astrbot/_internal/abc/base_astrbot_orchestrator.py b/astrbot/_internal/abc/base_astrbot_orchestrator.py deleted file mode 100644 index c6c17ee89..000000000 --- a/astrbot/_internal/abc/base_astrbot_orchestrator.py +++ /dev/null @@ -1,352 +0,0 @@ -""" -AstrBot Orchestrator - core runtime lifecycle manager. - -Architecture -============ - - ┌─────────────────────────────────────────────────────┐ - │ Orchestrator │ - │ (owns lifecycle of all protocol clients + stars) │ - └─────────────────────────────────────────────────────┘ - │ - ┌──────────────┼──────────────┐ - ▼ ▼ ▼ - ┌─────────┐ ┌─────────┐ ┌─────────┐ - │ LSP │ │ MCP │ │ ACP │ - │ Client │ │ Client │ │ Client │ - └─────────┘ └─────────┘ └─────────┘ - │ │ │ - ▼ ▼ ▼ - LSP Servers MCP Servers ACP Services - - ┌─────────────────────────────────────────────────────┐ - │ ABP Client │ - │ (in-process star registry) │ - └─────────────────────────────────────────────────────┘ - │ - ▼ - ┌─────────┐ - │ Stars │ - │(Plugins) │ - └─────────┘ - - -Lifecycle State Machine -======================= - - States: - ┌─────────┐ - │ INIT │───► orchestrator created, clients not initialized - └────┬────┘ - │ start() - ▼ - ┌─────────┐ - │ RUNNING │◄─── run_loop() executing - └────┬────┘ - │ shutdown() - ▼ - ┌──────────┐ - │ SHUTDOWN │─── all clients closed, ready for GC - └──────────┘ - - Transitions: - INIT + start() ──► RUNNING - RUNNING + shutdown() ──► SHUTDOWN - - For each protocol client, the orchestrator: - 1. Creates instance in __init__ - 2. Calls connect() to initialize - 3. Calls protocol-specific setup (connect_to_server, etc) - 4. Manages via run_loop() heartbeat - 5. Calls shutdown() on final cleanup - - -Star Registration Flow -===================== - - orchestrator.register_star("my-star", MyStar()) - │ - ▼ - ┌───────────────────┐ - │ ABP Client │ - │ .register_star() │ - └───────────────────┘ - │ - ▼ - ┌───────────────────┐ - │ Internal dict │ - │ {"my-star": obj} │ - └───────────────────┘ - - -Message Routing (conceptual) -=========================== - - External Tool Call - │ - ▼ - ┌──────────────┐ list_tools() ┌──────────────┐ - │ MCP Client │────────────────────►│ MCP Server │ - └──────────────┘◄────────────────────└──────────────┘ - │ tool result - ▼ - ┌──────────────┐ call_tool() ┌──────────────┐ - │ ABP │────────────────────►│ Star │ - │ Client │◄────────────────────└──────────────┘ - └──────────────┘ tool result - │ - ▼ - Return to caller - - -run_loop() Responsibilities -=========================== - - while running: - │─ check LSP server health (ping/heartbeat) - │─ check MCP session status (reconnect if needed) - │─ check ACP client connections - │─ process any pending star notifications - │─ sleep(SLEEP_INTERVAL) - - -Shutdown Sequence -================== - - shutdown() - │ - ├─ set _running = False - │ - ├─ LSP.shutdown() - │ └─ send "shutdown" request - │ └─ terminate subprocess - │ - ├─ ACP.shutdown() - │ └─ close TCP/Unix connections - │ - ├─ ABP.shutdown() - │ └─ cancel pending requests - │ - └─ MCP.cleanup() - └─ close all sessions - └─ cleanup subprocesses - - -Exception Handling -================== - - Each protocol client should: - - Catch connection errors - - Attempt reconnection with exponential backoff - - Log errors but don't crash run_loop - - Raise on irrecoverable failures - - The orchestrator run_loop should: - - Catch CancelledError on shutdown - - Catch Exception and log (don't crash) - - Ensure cleanup runs in finally block -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: - from astrbot._internal.protocols.abp.client import AstrbotAbpClient - from astrbot._internal.protocols.acp.client import AstrbotAcpClient - from astrbot._internal.protocols.lsp.client import AstrbotLspClient - from astrbot._internal.protocols.mcp.client import McpClient - - -#: Default heartbeat interval for run_loop() -DEFAULT_SLEEP_INTERVAL: float = 5.0 - - -class BaseAstrbotOrchestrator(ABC): - """ - Core runtime: owns lifecycle of all protocol clients and stars. - - ┌────────────────────────────────────────────────────────────┐ - │ Protocol Clients (always present, never None after init) │ - ├────────────────────────────────────────────────────────────┤ - │ lsp: Language Server Protocol │ - │ Purpose: code completion, diagnostics, hover, etc │ - │ Transport: stdio subprocess │ - │ │ - │ mcp: Model Context Protocol │ - │ Purpose: external tool access │ - │ Transport: stdio | SSE | HTTP │ - │ │ - │ acp: AstrBot Communication Protocol │ - │ Purpose: inter-service communication │ - │ Transport: TCP | Unix Socket │ - │ │ - │ abp: AstrBot Protocol │ - │ Purpose: in-process star (plugin) communication │ - │ Transport: direct method calls │ - └────────────────────────────────────────────────────────────┘ - - ┌────────────────────────────────────────────────────────────┐ - │ Star Registry │ - ├────────────────────────────────────────────────────────────┤ - │ _stars: dict[str, Any] │ - │ Stars are plugins registered by name │ - │ ABP client delegates calls to registered stars │ - └────────────────────────────────────────────────────────────┘ - - Subclass must implement: - - __init__(): create all protocol client instances - - run_loop(): main event loop (block until shutdown) - - register_star(name, instance): add to registry + ABP - - unregister_star(name): remove from registry + ABP - - shutdown(): clean up all clients - """ - - #: LSP client for language intelligence - lsp: AstrbotLspClient - - #: MCP client for external tools - mcp: McpClient - - #: ACP client for inter-service communication - acp: AstrbotAcpClient - - #: ABP client for in-process star communication - abp: AstrbotAbpClient - - def __init__(self) -> None: - """ - Initialize orchestrator and all protocol clients. - - After __init__, all clients exist but are not connected. - Call start() or run_loop() to begin operation. - - Example: - class MyOrchestrator(BaseAstrbotOrchestrator): - def __init__(self): - self.lsp = AstrbotLspClient() - self.mcp = McpClient() - self.acp = AstrbotAcpClient() - self.abp = AstrbotAbpClient() - self._stars: dict[str, Any] = {} - self._running = False - """ - self._stars: dict[str, Any] = {} - self._running: bool = False - - @property - def running(self) -> bool: - """True if run_loop() is executing.""" - return self._running - - @abstractmethod - async def start(self) -> None: - """ - Initialize all protocol clients. - - Called once before run_loop(). Should: - 1. Call lsp.connect() - 2. Call mcp.connect() - 3. Call acp.connect() - 4. Call abp.connect() - 5. Set _running = True - - Raises: - Exception: if any client fails to initialize - """ - ... - - @abstractmethod - async def run_loop(self) -> None: - """ - Main event loop - blocks until shutdown. - - Execution: - self._running = True - try: - while self._running: - await self._heartbeat() - await anyio.sleep(DEFAULT_SLEEP_INTERVAL) - except asyncio.CancelledError: - pass # shutdown requested - finally: - self._running = False - - _heartbeat() responsibilities: - - Check LSP server health (optional ping) - - Check MCP session status, reconnect if needed - - Check ACP connections - - Process any pending star notifications - - Raises: - asyncio.CancelledError: when shutdown() called - - Note: - Subclass defines _heartbeat() for periodic tasks. - This method only handles the loop control. - """ - ... - - @abstractmethod - async def register_star(self, name: str, star_instance: Any) -> None: - """ - Register a star (plugin) with the orchestrator. - - Args: - name: Unique identifier for the star - instance: Star plugin instance (must have .call_tool() method) - - Does: - self._stars[name] = star_instance - self.abp.register_star(name, star_instance) - - Raises: - ValueError: if name already registered - """ - ... - - @abstractmethod - async def unregister_star(self, name: str) -> None: - """ - Unregister a star (plugin) from the orchestrator. - - Args: - name: Identifier of star to remove - - Does: - del self._stars[name] - self.abp.unregister_star(name) - - Note: - Idempotent - does nothing if name not found. - """ - ... - - @abstractmethod - async def get_star(self, name: str) -> Any | None: - """Get registered star by name. Returns None if not found.""" - ... - - @abstractmethod - async def list_stars(self) -> list[str]: - """Return list of registered star names.""" - ... - - @abstractmethod - async def shutdown(self) -> None: - """ - Graceful shutdown of orchestrator and all clients. - - Execution order: - 1. self._running = False (stop run_loop) - 2. await lsp.shutdown() - 3. await acp.shutdown() - 4. await abp.shutdown() - 5. await mcp.cleanup() - - Does NOT unregister stars - caller should do that first. - - After shutdown, orchestrator is ready for garbage collection. - """ - ... diff --git a/astrbot/_internal/abc/lsp/base_astrbot_lsp_client.py b/astrbot/_internal/abc/lsp/base_astrbot_lsp_client.py deleted file mode 100644 index 6aa38aace..000000000 --- a/astrbot/_internal/abc/lsp/base_astrbot_lsp_client.py +++ /dev/null @@ -1,114 +0,0 @@ -""" -LSP (Language Server Protocol) client. - -Transport: stdio subprocess -Messages: JSON-RPC 2.0 with Content-Length header -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: - pass - - -class LspMessage: - """JSON-RPC 2.0 message.""" - - jsonrpc: str = "2.0" - id: int | str | None = None - method: str | None = None - params: dict[str, Any] | None = None - result: Any = None - error: dict[str, Any] | None = None - - -class LspRequest(LspMessage): - """Outgoing request.""" - - def __init__(self, method: str, params: dict[str, Any] | None = None) -> None: - self.id = id(self) - self.method = method - self.params = params - - -class LspResponse(LspMessage): - """Incoming response.""" - - -class LspNotification(LspMessage): - """Incoming notification (no id).""" - - -class BaseAstrbotLspClient(ABC): - """ - LSP client: connects to LSP servers via stdio subprocess. - - Subclass must implement: - - connect() -> None - - connect_to_server(command, workspace_uri) -> None - - send_request(method, params) -> dict - - send_notification(method, params) -> None - - shutdown() -> None - """ - - @property - @abstractmethod - def connected(self) -> bool: - """True if connected to an LSP server.""" - ... - - @abstractmethod - async def connect(self) -> None: - self._connected = False - ... - - @abstractmethod - async def connect_to_server( - self, - command: list[str], - workspace_uri: str, - ) -> None: - """ - Start LSP server subprocess and complete handshake. - - Steps: - 1. Spawn subprocess with stdin/stdout pipes - 2. Send initialize request - 3. Wait for response - 4. Send initialized notification - """ - ... - - @abstractmethod - async def send_request( - self, - method: str, - params: dict[str, Any] | None = None, - ) -> Any: - """ - Send JSON-RPC request and return result. - - Raises: - RuntimeError: not connected - Exception: server returned error - """ - ... - - @abstractmethod - async def send_notification( - self, - method: str, - params: dict[str, Any] | None = None, - ) -> None: - """ - Send JSON-RPC notification (no response expected). - """ - ... - - @abstractmethod - async def shutdown(self) -> None: - """Send shutdown, terminate subprocess, cleanup.""" - ... diff --git a/astrbot/_internal/abc/mcp/base_astrbot_mcp_client.py b/astrbot/_internal/abc/mcp/base_astrbot_mcp_client.py deleted file mode 100644 index 091f704aa..000000000 --- a/astrbot/_internal/abc/mcp/base_astrbot_mcp_client.py +++ /dev/null @@ -1,95 +0,0 @@ -""" -MCP (Model Context Protocol) client. - -Transport: stdio | SSE | streamable_http -Messages: JSON-RPC 2.0 -""" - -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Any, Literal, TypedDict - -if TYPE_CHECKING: - pass - - -class McpServerConfig(TypedDict, total=False): - """MCP server configuration.""" - - # Stdio transport - command: str - args: list[str] - env: dict[str, str] - cwd: str - - # HTTP transport - url: str - headers: dict[str, str] - transport: Literal["sse", "streamable_http"] - - -class McpToolInfo(TypedDict): - """MCP tool descriptor.""" - - name: str - description: str - inputSchema: dict[str, Any] - - -class BaseAstrbotMcpClient(ABC): - """ - MCP client: connects to MCP servers for external tools. - - Subclass must implement: - - connect() -> None - - connect_to_server(config, name) -> None - - list_tools() -> list[McpToolInfo] - - call_tool(name, args, timeout) -> CallToolResult - - cleanup() -> None - """ - - session: Any # mcp.ClientSession - - @property - @abstractmethod - def connected(self) -> bool: ... - - @abstractmethod - async def connect(self) -> None: - """Initialize client session.""" - ... - - @abstractmethod - async def connect_to_server( - self, - config: McpServerConfig, - name: str, - ) -> None: - """ - Connect to MCP server. - - Stdio: {"command": "python", "args": ["server.py"], "env": {...}} - HTTP: {"url": "https://...", "transport": "sse"} - """ - ... - - @abstractmethod - async def list_tools(self) -> list[McpToolInfo]: - """Call tools/list and return tools.""" - ... - - @abstractmethod - async def call_tool( - self, - name: str, - arguments: dict[str, Any], - read_timeout_seconds: int = 60, - ) -> Any: - """Call tools/call with reconnection support.""" - ... - - @abstractmethod - async def cleanup(self) -> None: - """Close all server connections.""" - ... diff --git a/astrbot/_internal/geteway/__init__.py b/astrbot/_internal/geteway/__init__.py deleted file mode 100644 index b88ac0e3b..000000000 --- a/astrbot/_internal/geteway/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -"""Gateway module - FastAPI server for the dashboard backend.""" - -from .server import AstrbotGateway -from .ws_manager import WebSocketManager - -__all__ = ["AstrbotGateway", "WebSocketManager"] diff --git a/astrbot/_internal/geteway/deps.py b/astrbot/_internal/geteway/deps.py deleted file mode 100644 index 73e648216..000000000 --- a/astrbot/_internal/geteway/deps.py +++ /dev/null @@ -1,4 +0,0 @@ -""" -依赖注入 - -""" diff --git a/astrbot/_internal/geteway/routes/inspector.py b/astrbot/_internal/geteway/routes/inspector.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/astrbot/_internal/geteway/routes/memory.py b/astrbot/_internal/geteway/routes/memory.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/astrbot/_internal/geteway/routes/stats.py b/astrbot/_internal/geteway/routes/stats.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/astrbot/_internal/geteway/server.py b/astrbot/_internal/geteway/server.py deleted file mode 100644 index b41edd529..000000000 --- a/astrbot/_internal/geteway/server.py +++ /dev/null @@ -1,248 +0,0 @@ -""" -AstrBot Gateway - FastAPI server for the dashboard backend. - -Provides REST API endpoints and WebSocket connections for the frontend dashboard. -The gateway acts as the communication bridge between the dashboard and the orchestrator. -""" - -from __future__ import annotations - -import json -from contextlib import asynccontextmanager -from typing import TYPE_CHECKING, Any, cast - -from astrbot import logger -from astrbot._internal.abc.base_astrbot_gateway import BaseAstrbotGateway -from astrbot._internal.abc.base_astrbot_orchestrator import BaseAstrbotOrchestrator -from astrbot._internal.geteway.ws_manager import WebSocketManager - -if TYPE_CHECKING: - from fastapi import FastAPI, WebSocket, WebSocketDisconnect -else: - try: - from fastapi import FastAPI, WebSocket, WebSocketDisconnect - from fastapi.middleware.cors import CORSMiddleware - except ImportError: - logger.warning("FastAPI not installed, gateway unavailable.") - FastAPI = cast(Any, None) - WebSocket = cast(Any, None) - WebSocketDisconnect = cast(Any, None) - CORSMiddleware = cast(Any, None) - -log = logger - - -class AstrbotGateway(BaseAstrbotGateway): - """ - FastAPI-based gateway server for AstrBot. - - Handles: - - REST API endpoints for configuration and stats - - WebSocket connections for real-time communication - - CORS middleware for dashboard access - """ - - def __init__(self, orchestrator: BaseAstrbotOrchestrator) -> None: - self.orchestrator = orchestrator - self.ws_manager = WebSocketManager() - self._app: FastAPI | None = None - self._host = "0.0.0.0" - self._port = 8765 - - async def serve(self) -> None: - """ - Start the gateway server. - - Creates and runs a FastAPI application with WebSocket support. - """ - if FastAPI is None: - raise RuntimeError("FastAPI is not installed") - - log.info(f"Starting AstrBot Gateway on {self._host}:{self._port}") - - @asynccontextmanager - async def lifespan(app: FastAPI): - # Startup - log.info("Gateway server started.") - yield - # Shutdown - await self.ws_manager.broadcast({"type": "server_shutdown"}) - log.info("Gateway server stopped.") - - self._app = FastAPI( - title="AstrBot Gateway", - description="Backend API for AstrBot dashboard", - version="1.0.0", - lifespan=lifespan, - ) - - # CORS middleware - self._app.add_middleware( - cast(Any, CORSMiddleware), - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) - - # Include routers - self._setup_routes() - - # Run with uvicorn - import uvicorn - - config = uvicorn.Config( - self._app, - host=self._host, - port=self._port, - log_level="info", - ) - server = uvicorn.Server(config) - await server.serve() - - def _setup_routes(self) -> None: - """Set up API routes.""" - if self._app is None: - return - - from fastapi import APIRouter - - # Health check - @self._app.get("/health") - async def health(): - return {"status": "ok"} - - # WebSocket endpoint - @self._app.websocket("/ws") - async def websocket_endpoint(ws: WebSocket): - await self.ws_manager.connect(ws) - try: - while True: - data = await ws.receive_text() - try: - message = json.loads(data) - response = await self._handle_ws_message(message) - if response: - await ws.send_json(response) - except json.JSONDecodeError: - await ws.send_json({"error": "Invalid JSON"}) - except WebSocketDisconnect: - self.ws_manager.disconnect(ws) - - # Stats router - stats_router = APIRouter(prefix="/api/stats", tags=["stats"]) - - @stats_router.get("/overview") - async def get_overview(): - return await self._get_stats_overview() - - self._app.include_router(stats_router) - - # Inspector router - inspector_router = APIRouter(prefix="/api/inspector", tags=["inspector"]) - - @inspector_router.get("/stars") - async def list_stars(): - return await self._list_stars() - - @inspector_router.get("/stars/{star_name}") - async def get_star(star_name: str): - return await self._get_star_detail(star_name) - - self._app.include_router(inspector_router) - - # Memory router - memory_router = APIRouter(prefix="/api/memory", tags=["memory"]) - - @memory_router.get("/") - async def get_memory(): - return await self._get_memory_info() - - self._app.include_router(memory_router) - - async def _handle_ws_message( - self, message: dict[str, Any] - ) -> dict[str, Any] | None: - """ - Handle an incoming WebSocket message. - - Args: - message: Parsed JSON message from the client - - Returns: - Response message to send back, or None for no response - """ - msg_type = message.get("type") - data = message.get("data", {}) - - if msg_type == "ping": - return {"type": "pong", "data": {}} - - if msg_type == "call_tool": - return await self._handle_call_tool(data) - - if msg_type == "get_stars": - return {"type": "stars_list", "data": await self._list_stars()} - - return { - "type": "error", - "data": {"message": f"Unknown message type: {msg_type}"}, - } - - async def _handle_call_tool(self, data: dict[str, Any]) -> dict[str, Any]: - """Handle a tool call request via WebSocket.""" - star_name = data.get("star") - tool_name = data.get("tool") - arguments = data.get("arguments", {}) - - if not star_name or not tool_name: - return { - "type": "tool_result", - "data": {"error": "Missing star or tool name"}, - } - - try: - result = await self.orchestrator.abp.call_star_tool( - star_name, tool_name, arguments - ) - return {"type": "tool_result", "data": {"result": result}} - except Exception as e: - return {"type": "tool_result", "data": {"error": str(e)}} - - async def _get_stats_overview(self) -> dict[str, Any]: - """Get overview statistics.""" - return { - "stars_count": len(self.orchestrator.abp._stars), - "lsp_connected": self.orchestrator.lsp._connected, - "mcp_sessions": getattr(self.orchestrator.mcp, "session", None) is not None, - "acp_clients": len(getattr(self.orchestrator.acp, "_clients", [])), - } - - async def _list_stars(self) -> list[dict[str, Any]]: - """List all registered stars.""" - stars = [] - for name in self.orchestrator.abp._stars: - stars.append({"name": name, "status": "active"}) - return stars - - async def _get_star_detail(self, star_name: str) -> dict[str, Any]: - """Get details of a specific star.""" - star = self.orchestrator.abp._stars.get(star_name) - if not star: - return {"error": f"Star '{star_name}' not found"} - return {"name": star_name, "status": "active"} - - async def _get_memory_info(self) -> dict[str, Any]: - """Get memory usage information.""" - import gc - - gc.collect() - return { - "gc_objects": len(gc.get_objects()), - "python_memory": "N/A", # Would need psutil for actual values - } - - def set_listen_address(self, host: str, port: int) -> None: - """Set the listen address for the gateway server.""" - self._host = host - self._port = port diff --git a/astrbot/_internal/geteway/ws_manager.py b/astrbot/_internal/geteway/ws_manager.py deleted file mode 100644 index d06510da7..000000000 --- a/astrbot/_internal/geteway/ws_manager.py +++ /dev/null @@ -1,103 +0,0 @@ -""" -WebSocket connection manager for the AstrBot gateway. -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING, Any, cast - -import anyio - -from astrbot import logger - -if TYPE_CHECKING: - from fastapi import WebSocket -else: - try: - from fastapi import WebSocket - except ImportError: - logger.warning("FastAPI not installed, WebSocketManager unavailable.") - WebSocket = cast(Any, None) - -log = logger - - -class WebSocketManager: - """ - Manages all active WebSocket connections. - - Provides connection/disconnection handling and broadcast capabilities. - """ - - def __init__(self) -> None: - self._connections: set[WebSocket] = set() - self._lock = anyio.Lock() - - async def connect(self, websocket: WebSocket) -> None: - """Accept and register a new WebSocket connection.""" - await websocket.accept() - async with self._lock: - self._connections.add(websocket) - log.debug(f"WebSocket connected. Total: {len(self._connections)}") - - async def disconnect(self, websocket: WebSocket) -> None: - """Remove a WebSocket connection.""" - async with self._lock: - self._connections.discard(websocket) - log.debug(f"WebSocket disconnected. Total: {len(self._connections)}") - - async def send_json(self, websocket: WebSocket, data: dict[str, Any]) -> None: - """ - Send JSON data to a specific WebSocket. - - Args: - websocket: Target WebSocket connection - data: Data to send (must be JSON-serializable) - """ - try: - await websocket.send_json(data) - except Exception as e: - log.warning(f"Failed to send to WebSocket: {e}") - await self.disconnect(websocket) - - async def broadcast(self, data: dict[str, Any]) -> None: - """ - Broadcast JSON data to all connected WebSockets. - - Args: - data: Data to broadcast (must be JSON-serializable) - """ - async with self._lock: - connections = list(self._connections) - - for conn in connections: - try: - await conn.send_json(data) - except Exception as e: - log.warning(f"Failed to broadcast to WebSocket: {e}") - async with self._lock: - self._connections.discard(conn) - - async def send_to( - self, websocket: WebSocket, message: str | dict[str, Any] - ) -> None: - """ - Send a message to a specific WebSocket. - - Args: - websocket: Target WebSocket connection - message: Message to send (string or dict) - """ - try: - if isinstance(message, str): - await websocket.send_text(message) - else: - await websocket.send_json(message) - except Exception as e: - log.warning(f"Failed to send to WebSocket: {e}") - await self.disconnect(websocket) - - @property - def connection_count(self) -> int: - """Return the number of active connections.""" - return len(self._connections) diff --git a/astrbot/_internal/protocols/abp/__init__.py b/astrbot/_internal/protocols/abp/__init__.py deleted file mode 100644 index 54f74818f..000000000 --- a/astrbot/_internal/protocols/abp/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""ABP module - AstrBot Protocol client implementation (built-in plugin protocol).""" - -from .client import AstrbotAbpClient - -__all__ = ["AstrbotAbpClient"] diff --git a/astrbot/_internal/protocols/abp/client.py b/astrbot/_internal/protocols/abp/client.py deleted file mode 100644 index 4c0725829..000000000 --- a/astrbot/_internal/protocols/abp/client.py +++ /dev/null @@ -1,93 +0,0 @@ -""" -ABP (AstrBot Protocol) client implementation. - -ABP is the built-in plugin protocol where the orchestrator acts as client -connecting to internal stars (plugins) embedded in the runtime. -""" - -from __future__ import annotations - -from typing import Any - -from astrbot import logger -from astrbot._internal.abc.abp.base_astrbot_abp_client import BaseAstrbotAbpClient - -log = logger - - -class AstrbotAbpClient(BaseAstrbotAbpClient): - """ - ABP client for communicating with internal stars (built-in plugins). - - The orchestrator acts as the client, sending requests to and receiving - notifications from stars running within the same process. - """ - - def __init__(self) -> None: - self._connected = False - self._stars: dict[str, Any] = {} - # Use a simple dict for pending requests; we avoid asyncio.Future here. - self._pending_requests: dict[str, Any] = {} - self._request_id = 0 - - @property - def connected(self) -> bool: - """True if connected to stars registry.""" - return self._connected - - async def connect(self) -> None: - """Connect to internal stars registry.""" - log.debug("ABP client connecting to internal stars...") - self._connected = True - log.info("ABP client connected to internal stars registry.") - - async def call_star_tool( - self, star_name: str, tool_name: str, arguments: dict[str, Any] - ) -> Any: - """ - Call a tool on a registered star. - - Args: - star_name: Name of the star (plugin) - tool_name: Name of the tool to call - arguments: Tool arguments - - Returns: - Tool call result - """ - if not self._connected: - raise RuntimeError("ABP client is not connected") - - star = self._stars.get(star_name) - if not star: - raise ValueError(f"Star '{star_name}' not found") - - request_id = f"{self._request_id}" - self._request_id += 1 - - # No asyncio.Future used; store a placeholder entry for tracking if needed. - self._pending_requests[request_id] = None - - try: - # Call the star's tool handler - result = await star.call_tool(tool_name, arguments) - return result - finally: - self._pending_requests.pop(request_id, None) - - def register_star(self, star_name: str, star_instance: Any) -> None: - """Register a star (plugin) with the ABP client.""" - self._stars[star_name] = star_instance - log.debug(f"Star '{star_name}' registered with ABP client.") - - def unregister_star(self, star_name: str) -> None: - """Unregister a star from the ABP client.""" - self._stars.pop(star_name, None) - log.debug(f"Star '{star_name}' unregistered from ABP client.") - - async def shutdown(self) -> None: - """Shutdown the ABP client connection.""" - self._connected = False - # Clear any pending requests (no asyncio futures used in this implementation) - self._pending_requests.clear() - log.info("ABP client shut down.") diff --git a/astrbot/_internal/protocols/acp/__init__.py b/astrbot/_internal/protocols/acp/__init__.py deleted file mode 100644 index 853768409..000000000 --- a/astrbot/_internal/protocols/acp/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -"""ACP module - AstrBot Communication Protocol client and server implementations.""" - -from .client import AstrbotAcpClient -from .server import AstrbotAcpServer - -__all__ = ["AstrbotAcpClient", "AstrbotAcpServer"] diff --git a/astrbot/_internal/protocols/acp/client.py b/astrbot/_internal/protocols/acp/client.py deleted file mode 100644 index 6eaf1e316..000000000 --- a/astrbot/_internal/protocols/acp/client.py +++ /dev/null @@ -1,220 +0,0 @@ -""" -ACP (AstrBot Communication Protocol) client implementation. - -ACP is a client-server protocol for inter-service communication, -similar to MCP but designed specifically for AstrBot's architecture. -""" - -from __future__ import annotations - -import asyncio -import json -from typing import Any - -from astrbot import logger -from astrbot._internal.abc.acp.base_astrbot_acp_client import BaseAstrbotAcpClient - -log = logger - - -class AstrbotAcpClient(BaseAstrbotAcpClient): - """ - ACP client for communicating with ACP servers. - - The orchestrator acts as an ACP client, connecting to external - ACP-compatible services. - """ - - def __init__(self) -> None: - self._connected = False - self._reader: asyncio.StreamReader | None = None - self._writer: asyncio.StreamWriter | None = None - self._server_url: str | None = None - self._pending_requests: dict[str, asyncio.Future[dict[str, Any]]] = {} - self._request_id = 0 - self._reader_task: asyncio.Task[None] | None = None - - @property - def connected(self) -> bool: - """True if connected to an ACP server.""" - return self._connected - - async def connect(self) -> None: - """ - Connect to configured ACP servers. - - ACP servers can be accessed via TCP (host:port) or Unix socket. - """ - log.debug("ACP client connecting...") - # TODO: Load ACP server configurations - self._connected = True - log.info("ACP client initialized.") - - async def connect_to_server(self, host: str, port: int) -> None: - """ - Connect to an ACP server via TCP. - - Args: - host: Server hostname or IP - port: Server port - """ - self._server_url = f"{host}:{port}" - self._reader, self._writer = await asyncio.open_connection(host, port) - self._connected = True - - # Start reading responses - self._reader_task = asyncio.create_task(self._read_messages()) - - log.info(f"ACP client connected to {self._server_url}") - - async def connect_to_unix_socket(self, socket_path: str) -> None: - """ - Connect to an ACP server via Unix socket. - - Args: - socket_path: Path to the Unix socket - """ - self._server_url = f"unix://{socket_path}" - self._reader, self._writer = await asyncio.open_unix_connection(socket_path) - self._connected = True - - self._reader_task = asyncio.create_task(self._read_messages()) - - log.info(f"ACP client connected to {self._server_url}") - - async def _read_messages(self) -> None: - """Background task to read ACP messages.""" - if not self._reader: - return - - buffer = b"" - while self._connected: - try: - data = await self._reader.read(4096) - if not data: - break - buffer += data - - while True: - header_end = buffer.find(b"\n") - if header_end == -1: - break - - try: - header = json.loads(buffer[:header_end].decode("utf-8")) - except json.JSONDecodeError: - buffer = buffer[header_end + 1 :] - continue - - content_length = header.get("content-length", 0) - if ( - content_length == 0 - or len(buffer) < header_end + 1 + content_length - ): - break - - content = buffer[header_end + 1 : header_end + 1 + content_length] - buffer = buffer[header_end + 1 + content_length :] - - message = json.loads(content.decode("utf-8")) - - if "id" in message: - request_id = str(message["id"]) - future = self._pending_requests.pop(request_id, None) - if future and not future.done(): - if "error" in message: - future.set_exception(Exception(str(message["error"]))) - else: - future.set_result(message.get("result", {})) - else: - await self._handle_notification(message) - - except Exception as e: - if self._connected: - log.error(f"ACP read error: {e}") - break - - async def _handle_notification(self, notification: dict[str, Any]) -> None: - """Handle incoming ACP notifications.""" - method = notification.get("method", "") - log.debug(f"ACP notification: {method}") - - async def call_tool( - self, server_name: str, tool_name: str, arguments: dict[str, Any] - ) -> Any: - """ - Call a tool on an ACP server. - - Args: - server_name: Name of the ACP server - tool_name: Name of the tool to call - arguments: Tool arguments - - Returns: - Tool call result - """ - if not self._connected: - raise RuntimeError("ACP client is not connected") - - request_id = str(self._request_id) - self._request_id += 1 - - message = { - "jsonrpc": "2.0", - "id": request_id, - "method": f"{server_name}/{tool_name}", - "params": arguments, - } - - future: asyncio.Future[dict[str, Any]] = asyncio.Future() - self._pending_requests[request_id] = future - - await self._send_message(message) - return await future - - async def _send_message(self, message: dict[str, Any]) -> None: - """Send an ACP message.""" - if not self._writer: - raise RuntimeError("ACP client not connected") - - content = json.dumps(message) - header = json.dumps({"content-length": len(content)}) + "\n" - - self._writer.write((header + content).encode()) - await self._writer.drain() - - async def send_notification( - self, method: str, params: dict[str, Any] | None = None - ) -> None: - """Send a one-way notification to the server.""" - message = { - "jsonrpc": "2.0", - "method": method, - "params": params or {}, - } - await self._send_message(message) - - async def shutdown(self) -> None: - """Shutdown the ACP client connection.""" - self._connected = False - - if self._reader_task: - self._reader_task.cancel() - try: - await self._reader_task - except asyncio.CancelledError: - pass - - if self._writer: - self._writer.close() - try: - await self._writer.wait_closed() - except Exception: - pass - - for future in self._pending_requests.values(): - if not future.done(): - future.cancel() - self._pending_requests.clear() - - log.info("ACP client shut down.") diff --git a/astrbot/_internal/protocols/acp/server.py b/astrbot/_internal/protocols/acp/server.py deleted file mode 100644 index 9f3159b45..000000000 --- a/astrbot/_internal/protocols/acp/server.py +++ /dev/null @@ -1,223 +0,0 @@ -""" -ACP (AstrBot Communication Protocol) server implementation. - -ACP servers listen for connections from ACP clients and provide -services/tools to the orchestrator. -""" - -from __future__ import annotations - -import asyncio -import json -from collections.abc import Callable -from typing import Any - -from astrbot import logger -from astrbot._internal.abc.acp.base_astrbot_acp_server import BaseAstrbotAcpServer - -log = logger - - -class AstrbotAcpServer(BaseAstrbotAcpServer): - """ - ACP server for accepting connections from ACP clients. - - ACP servers expose tools/notifications that can be called by clients. - """ - - def __init__(self) -> None: - self._running = False - self._host: str = "127.0.0.1" - self._port: int = 8765 - self._server: asyncio.Server | None = None - self._clients: set[tuple[asyncio.StreamReader, asyncio.StreamWriter]] = set() - self._tool_handlers: dict[str, Callable[..., Any]] = {} - self._notification_handlers: dict[str, Callable[..., Any]] = {} - - def register_tool(self, name: str, handler: Callable[..., Any]) -> None: - """ - Register a tool handler. - - Args: - name: Tool name - handler: Async callable that handles tool calls - """ - self._tool_handlers[name] = handler - log.debug(f"ACP server registered tool: {name}") - - def register_notification_handler( - self, name: str, handler: Callable[..., Any] - ) -> None: - """ - Register a notification handler. - - Args: - name: Notification method name - handler: Async callable that handles notifications - """ - self._notification_handlers[name] = handler - log.debug(f"ACP server registered notification handler: {name}") - - async def start(self, host: str = "127.0.0.1", port: int = 8765) -> None: - """ - Start the ACP server. - - Args: - host: Host to bind to - port: Port to listen on - """ - self._host = host - self._port = port - self._server = await asyncio.start_server( - self._handle_client, - host=host, - port=port, - ) - self._running = True - log.info(f"ACP server listening on {host}:{port}") - - async def _handle_client( - self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter - ) -> None: - """Handle an incoming ACP client connection.""" - addr = writer.get_extra_info("peername") - log.debug(f"ACP client connected: {addr}") - self._clients.add((reader, writer)) - - buffer = b"" - try: - while self._running: - try: - data = await reader.read(4096) - if not data: - break - buffer += data - - while True: - header_end = buffer.find(b"\n") - if header_end == -1: - break - - try: - header = json.loads(buffer[:header_end].decode("utf-8")) - except json.JSONDecodeError: - buffer = buffer[header_end + 1 :] - continue - - content_length = header.get("content-length", 0) - if ( - content_length == 0 - or len(buffer) < header_end + 1 + content_length - ): - break - - content = buffer[ - header_end + 1 : header_end + 1 + content_length - ] - buffer = buffer[header_end + 1 + content_length :] - - message = json.loads(content.decode("utf-8")) - response = await self._handle_message(message) - - if response: - content = json.dumps(response) - resp_header = ( - json.dumps({"content-length": len(content)}) + "\n" - ) - writer.write(resp_header.encode() + content.encode()) - await writer.drain() - - except Exception as e: - log.error(f"ACP client error ({addr}): {e}") - break - - finally: - self._clients.discard((reader, writer)) - writer.close() - try: - await writer.wait_closed() - except Exception: - pass - log.debug(f"ACP client disconnected: {addr}") - - async def _handle_message(self, message: dict[str, Any]) -> dict[str, Any] | None: - """Handle an incoming ACP message.""" - method = message.get("method", "") - msg_id = message.get("id") - params = message.get("params", {}) - - # Check if it's a notification (no id) or request (has id) - if msg_id is None: - # Notification - handler = self._notification_handlers.get(method) - if handler: - try: - await handler(params) - except Exception as e: - log.error(f"ACP notification handler error ({method}): {e}") - return None - - # Request - result = None - error = None - - handler = self._tool_handlers.get(method) - if handler: - try: - result = await handler(params) - except Exception as e: - error = str(e) - log.error(f"ACP tool handler error ({method}): {e}") - else: - error = f"Unknown method: {method}" - - response: dict[str, Any] = {"jsonrpc": "2.0", "id": msg_id} - if error: - response["error"] = {"code": -32601, "message": error} - else: - response["result"] = result - - return response - - async def broadcast_notification(self, method: str, params: dict[str, Any]) -> None: - """ - Broadcast a notification to all connected clients. - - Args: - method: Notification method name - params: Notification parameters - """ - message = { - "jsonrpc": "2.0", - "method": method, - "params": params, - } - content = json.dumps(message) - header = json.dumps({"content-length": len(content)}) + "\n" - data = header.encode() + content.encode() - - for reader, writer in list(self._clients): - try: - writer.write(data) - await writer.drain() - except Exception as e: - log.warning(f"Failed to broadcast to client: {e}") - - async def shutdown(self) -> None: - """Shutdown the ACP server.""" - self._running = False - - if self._server: - self._server.close() - await self._server.wait_closed() - self._server = None - - for reader, writer in list(self._clients): - writer.close() - try: - await writer.wait_closed() - except Exception: - pass - self._clients.clear() - - log.info("ACP server shut down.") diff --git a/astrbot/_internal/protocols/lsp/__init__.py b/astrbot/_internal/protocols/lsp/__init__.py deleted file mode 100644 index f7708a27d..000000000 --- a/astrbot/_internal/protocols/lsp/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""LSP module - Language Server Protocol client implementation.""" - -from .client import AstrbotLspClient - -__all__ = ["AstrbotLspClient"] diff --git a/astrbot/_internal/protocols/lsp/client.py b/astrbot/_internal/protocols/lsp/client.py deleted file mode 100644 index 21b12e981..000000000 --- a/astrbot/_internal/protocols/lsp/client.py +++ /dev/null @@ -1,263 +0,0 @@ -""" -LSP (Language Server Protocol) client implementation. - -The orchestrator acts as an LSP client, connecting to LSP servers -that provide language intelligence features (completions, diagnostics, etc.). -""" - -from __future__ import annotations - -import asyncio -import json -from typing import Any - -import anyio -from anyio.abc import ByteReceiveStream, ByteSendStream, Process - -from astrbot import logger -from astrbot._internal.abc.lsp.base_astrbot_lsp_client import BaseAstrbotLspClient - -log = logger - - -class AstrbotLspClient(BaseAstrbotLspClient): - """ - LSP client for communicating with LSP servers. - - Implements the Microsoft Language Server Protocol for connecting to - external language intelligence services. - """ - - def __init__(self) -> None: - self._connected = False - self._reader: ByteReceiveStream | None = None - self._writer: ByteSendStream | None = None - self._server_process: Process | None = None - self._pending_requests: dict[int, Any] = {} - self._request_id = 0 - self._server_command: list[str] | None = None - self._reader_task: asyncio.Task[None] | None = None - - @property - def connected(self) -> bool: - """True if connected to an LSP server.""" - return self._connected - - async def _stop_reader_task(self) -> None: - reader_task = self._reader_task - if reader_task is None: - return - self._reader_task = None - if reader_task is asyncio.current_task(): - return - if not reader_task.done(): - reader_task.cancel() - try: - await reader_task - except asyncio.CancelledError: - pass - except Exception as exc: - log.debug("Ignoring failed LSP reader task during teardown", exc_info=exc) - - async def connect(self) -> None: - """ - Connect to configured LSP servers. - - LSP servers are typically stdio-based subprocesses. This method - establishes the communication channel. - """ - log.debug("LSP client connecting...") - # TODO: Load LSP server configurations and start subprocesses - # For now, mark as connected in idle mode - self._connected = True - log.info("LSP client initialized.") - - async def connect_to_server(self, command: list[str], workspace_uri: str) -> None: - """ - Connect to an LSP server subprocess. - - Args: - command: Command line to start the LSP server (e.g., ["python", "lsp_server.py"]) - workspace_uri: Root URI of the workspace to serve - """ - log.debug(f"Starting LSP server: {' '.join(command)}") - - await self._stop_reader_task() - - self._server_process = await anyio.open_process( - command, - stdin=-1, - stdout=-1, - stderr=-1, - ) - self._reader = self._server_process.stdout - self._writer = self._server_process.stdin - self._server_command = command - self._connected = True - - # Start reading responses in the background. - self._reader_task = asyncio.create_task(self._read_responses()) - - # Send initialize request - await self.send_request( - "initialize", - { - "processId": None, - "rootUri": workspace_uri, - "capabilities": {}, - }, - ) - - # Send initialized notification - await self.send_notification("initialized", {}) - - log.info(f"LSP client connected to server: {command[0]}") - - async def send_request( - self, method: str, params: dict[str, Any] | None = None - ) -> Any: - """Send an LSP request and wait for response.""" - if not self._writer: - raise RuntimeError("LSP client not connected") - - request_id = self._request_id - self._request_id += 1 - - message = { - "jsonrpc": "2.0", - "id": request_id, - "method": method, - "params": params or {}, - } - - # Use anyio.Event for request/response matching - response_event: anyio.Event = anyio.Event() - response_holder: dict[str, Any] = {} - - async def set_response(response: dict[str, Any]) -> None: - response_holder["response"] = response - response_event.set() - - self._pending_requests[request_id] = set_response - - content = json.dumps(message) - headers = f"Content-Length: {len(content)}\r\n\r\n" - await self._writer.send((headers + content).encode()) - - # Wait for response with timeout - with anyio.move_on_after(30): - await response_event.wait() - - if "response" in response_holder: - return response_holder["response"] - raise TimeoutError(f"LSP request {method} timed out") - - async def send_notification( - self, method: str, params: dict[str, Any] | None = None - ) -> None: - """Send an LSP notification (no response expected).""" - if not self._writer: - raise RuntimeError("LSP client not connected") - - message = { - "jsonrpc": "2.0", - "method": method, - "params": params or {}, - } - - content = json.dumps(message) - headers = f"Content-Length: {len(content)}\r\n\r\n" - await self._writer.send((headers + content).encode()) - - async def _read_responses(self) -> None: - """Background task to read LSP responses.""" - if not self._reader: - return - - buffer = b"" - try: - while self._connected: - try: - data = await self._reader.receive() - if not data: - break - buffer += data - - while True: - # Parse Content-Length header - header_end = buffer.find(b"\r\n\r\n") - if header_end == -1: - break - - header = buffer[:header_end].decode("utf-8") - content_length = 0 - for line in header.split("\r\n"): - if line.startswith("Content-Length:"): - content_length = int(line.split(":")[1].strip()) - - if content_length == 0: - break - - total_length = header_end + 4 + content_length - if len(buffer) < total_length: - break - - content = buffer[header_end + 4 : total_length] - buffer = buffer[total_length:] - - response = json.loads(content.decode("utf-8")) - - # Handle response vs notification - if "id" in response: - request_id = response["id"] - handler = self._pending_requests.pop(request_id, None) - if handler: - await handler(response) - else: - # Notification (e.g., window/logMessage) - await self._handle_notification(response) - - except anyio.EndOfStream: - break - except asyncio.CancelledError: - raise - except Exception as exc: - if self._connected: - self._connected = False - log.error("LSP reader task failed", exc_info=exc) - return - else: - if self._connected: - self._connected = False - log.warning("LSP reader task exited unexpectedly") - finally: - if self._reader_task is asyncio.current_task(): - self._reader_task = None - - async def _handle_notification(self, notification: dict[str, Any]) -> None: - """Handle incoming LSP notifications.""" - method = notification.get("method", "") - log.debug(f"LSP notification: {method}") - - async def shutdown(self) -> None: - """Shutdown the LSP client.""" - self._connected = False - - await self._stop_reader_task() - - if self._server_process: - try: - await self.send_notification("shutdown", {}) - except Exception: - pass - - self._server_process.terminate() - try: - with anyio.move_on_after(5): - await self._server_process.wait() - except Exception: - self._server_process.kill() - self._server_process = None - - self._pending_requests.clear() - log.info("LSP client shut down.") diff --git a/astrbot/_internal/protocols/mcp/__init__.py b/astrbot/_internal/protocols/mcp/__init__.py deleted file mode 100644 index 7826f38f4..000000000 --- a/astrbot/_internal/protocols/mcp/__init__.py +++ /dev/null @@ -1,63 +0,0 @@ -"""MCP module - Model Context Protocol client and tool implementations. - -This module provides MCP client functionality and MCP tool wrappers. -""" - -import asyncio -from dataclasses import dataclass - -from .client import McpClient -from .config import ( - DEFAULT_MCP_CONFIG, - get_mcp_config_path, - load_mcp_config, - save_mcp_config, -) -from .tool import MCPTool - - -# Exceptions -class MCPInitError(Exception): - """Base exception for MCP initialization failures.""" - - -class MCPInitTimeoutError(asyncio.TimeoutError, MCPInitError): - """Raised when MCP client initialization exceeds the configured timeout.""" - - -class MCPAllServicesFailedError(MCPInitError): - """Raised when all configured MCP services fail to initialize.""" - - -class MCPShutdownTimeoutError(asyncio.TimeoutError): - """Raised when MCP shutdown exceeds the configured timeout.""" - - def __init__(self, names: list[str], timeout: float) -> None: - self.names = names - self.timeout = timeout - message = f"MCP 服务关闭超时({timeout:g} 秒):{', '.join(names)}" - super().__init__(message) - - -@dataclass -class MCPInitSummary: - """Summary of MCP initialization results.""" - - total: int - success: int - failed: list[str] - - -__all__ = [ - "DEFAULT_MCP_CONFIG", - "MCPAllServicesFailedError", - "MCPInitError", - "MCPInitSummary", - "MCPInitTimeoutError", - "MCPShutdownTimeoutError", - "MCPTool", - "McpClient", - "get_mcp_config_path", - "load_mcp_config", - "save_mcp_config", -] diff --git a/astrbot/_internal/protocols/mcp/client.py b/astrbot/_internal/protocols/mcp/client.py deleted file mode 100644 index 1badf9e6c..000000000 --- a/astrbot/_internal/protocols/mcp/client.py +++ /dev/null @@ -1,486 +0,0 @@ -"""MCP client implementation.""" - -import asyncio -import logging -import os -import sys -from contextlib import AsyncExitStack -from datetime import timedelta -from typing import Any, cast - -from tenacity import ( - before_sleep_log, - retry, - retry_if_exception_type, - stop_after_attempt, - wait_exponential, -) - -from astrbot._internal.abc.mcp.base_astrbot_mcp_client import ( - BaseAstrbotMcpClient, - McpServerConfig, - McpToolInfo, -) -from astrbot.core.utils.log_pipe import LogPipe - -logger = logging.getLogger("astrbot") - - -try: - import anyio - - import mcp - from mcp.client.sse import sse_client -except (ModuleNotFoundError, ImportError): - logger.warning( - "Warning: Missing 'mcp' dependency, MCP services will be unavailable." - ) - -try: - from mcp.client.streamable_http import streamablehttp_client -except (ModuleNotFoundError, ImportError): - logger.warning( - "Warning: Missing 'mcp' dependency or MCP library version too old, Streamable HTTP connection unavailable.", - ) - - -class TenacityLogger: - """Wraps a logging.Logger to satisfy tenacity's LoggerProtocol.""" - - __slots__ = ("_logger",) - _logger: logging.Logger - - def __init__(self, logger: logging.Logger) -> None: - self._logger = logger - - def log( - self, - level: int, - msg: str, - /, - *args: Any, - **kwargs: Any, - ) -> None: - self._logger.log(level, msg, *args, **kwargs) - - -def _prepare_config(config: dict) -> dict: - """Prepare configuration, handle nested format.""" - if config.get("mcpServers"): - first_key = next(iter(config["mcpServers"])) - config = config["mcpServers"][first_key] - config.pop("active", None) - return config - - -def _prepare_stdio_env(config: dict) -> dict: - """Preserve Windows executable resolution for stdio subprocesses.""" - if sys.platform != "win32": - return config - - pathext = os.environ.get("PATHEXT") - if not pathext: - return config - - prepared = config.copy() - env = dict(prepared.get("env") or {}) - env.setdefault("PATHEXT", pathext) - prepared["env"] = env - return prepared - - -async def _quick_test_mcp_connection(config: dict) -> tuple[bool, str]: - """Quick test MCP server connectivity.""" - import aiohttp - - cfg = _prepare_config(config.copy()) - - url = cfg["url"] - headers = cfg.get("headers", {}) - timeout = cfg.get("timeout", 10) - - try: - if "transport" in cfg: - transport_type = cfg["transport"] - elif "type" in cfg: - transport_type = cfg["type"] - else: - raise Exception("MCP connection config missing transport or type field") - - async with aiohttp.ClientSession() as session: - if transport_type == "streamable_http": - test_payload = { - "jsonrpc": "2.0", - "method": "initialize", - "id": 0, - "params": { - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": {"name": "test-client", "version": "1.2.3"}, - }, - } - async with session.post( - url, - headers={ - **headers, - "Content-Type": "application/json", - "Accept": "application/json, text/event-stream", - }, - json=test_payload, - timeout=aiohttp.ClientTimeout(total=timeout), - ) as response: - if response.status == 200: - return True, "" - return False, f"HTTP {response.status}: {response.reason}" - else: - async with session.get( - url, - headers={ - **headers, - "Accept": "application/json, text/event-stream", - }, - timeout=aiohttp.ClientTimeout(total=timeout), - ) as response: - if response.status == 200: - return True, "" - return False, f"HTTP {response.status}: {response.reason}" - - except asyncio.TimeoutError: - return False, f"Connection timeout: {timeout} seconds" - except Exception as e: - return False, f"{e!s}" - - -class McpClient(BaseAstrbotMcpClient): - def __init__(self) -> None: - # Initialize session and client objects - self.session: mcp.ClientSession | None = None - self.exit_stack = AsyncExitStack() - self._old_exit_stacks: list[AsyncExitStack] = [] # Track old stacks for cleanup - - self.name: str | None = None - self.active: bool = True - self.tools: list[mcp.Tool] = [] - self.server_errlogs: list[str] = [] - self.running_event = anyio.Event() - self.process_pid: int | None = None - - # Store connection config for reconnection - self._mcp_server_config: McpServerConfig | None = None - self._server_name: str | None = None - self._reconnect_lock = anyio.Lock() # Lock for thread-safe reconnection - self._reconnecting: bool = False # For logging and debugging - - async def connect(self) -> None: - """Initialize the MCP client connection. - - Note: Actual server connections are made via connect_to_server(). - This method prepares the client for use. - """ - # MCP client is initialized on-demand via connect_to_server - # This is a no-op stub to satisfy BaseAstrbotMcpClient - logger.debug("MCP client initialized.") - - @property - def connected(self) -> bool: - """True if MCP client has an active session.""" - return self.session is not None - - async def list_tools(self) -> list[McpToolInfo]: - """List all tools from connected MCP servers.""" - if not self.session: - return [] - result = await self.list_tools_and_save() - tools = [ - { - "name": tool.name, - "description": tool.description or "", - "inputSchema": tool.inputSchema, - } - for tool in result.tools - ] - return cast(list[McpToolInfo], tools) - - async def call_tool( - self, - name: str, - arguments: dict[str, Any], - read_timeout_seconds: int = 60, - ) -> Any: - """Call a tool on the MCP server with reconnection support.""" - return await self.call_tool_with_reconnect( - tool_name=name, - arguments=arguments, - read_timeout_seconds=timedelta(seconds=read_timeout_seconds), - ) - - @staticmethod - def _extract_stdio_process_pid(streams_context: object) -> int | None: - """Best-effort extraction for stdio subprocess PID used by lease cleanup. - - TODO(refactor): replace this async-generator frame introspection with a - stable MCP library hook once the upstream transport exposes process PID. - """ - generator = getattr(streams_context, "gen", None) - frame = getattr(generator, "ag_frame", None) - if frame is None: - return None - process = frame.f_locals.get("process") - pid = getattr(process, "pid", None) - try: - return int(pid) if pid is not None else None - except (TypeError, ValueError): - return None - - async def connect_to_server(self, config: McpServerConfig, name: str) -> None: - """Connect to MCP server - - If `url` parameter exists: - 1. When transport is specified as `streamable_http`, use Streamable HTTP connection. - 2. When transport is specified as `sse`, use SSE connection. - 3. If not specified, default to SSE connection to MCP service. - - Args: - config: Configuration for the MCP server. See https://modelcontextprotocol.io/quickstart/server - - """ - # Store config for reconnection - self._mcp_server_config = config - self._server_name = name - self.process_pid = None - - cfg = _prepare_config(dict(config)) - - def logging_callback( - msg: str | mcp.types.LoggingMessageNotificationParams, - ) -> None: - # Handle MCP service error logs - if isinstance(msg, mcp.types.LoggingMessageNotificationParams): - if msg.level in ("warning", "error", "critical", "alert", "emergency"): - log_msg = f"[{msg.level.upper()}] {msg.data!s}" - self.server_errlogs.append(log_msg) - - if "url" in cfg: - success, error_msg = await _quick_test_mcp_connection(cfg) - if not success: - raise Exception(error_msg) - - if "transport" in cfg: - transport_type = cfg["transport"] - elif "type" in cfg: - transport_type = cfg["type"] - else: - raise Exception("MCP connection config missing transport or type field") - - if transport_type != "streamable_http": - # SSE transport method - self._streams_context = sse_client( - url=cfg["url"], - headers=cfg.get("headers", {}), - timeout=cfg.get("timeout", 5), - sse_read_timeout=cfg.get("sse_read_timeout", 60 * 5), - ) - streams = await self.exit_stack.enter_async_context( - self._streams_context, - ) - - # Create a new client session - read_timeout = timedelta(seconds=cfg.get("session_read_timeout", 60)) - self.session = await self.exit_stack.enter_async_context( - mcp.ClientSession( - *streams, - read_timeout_seconds=read_timeout, - logging_callback=cast(Any, logging_callback), - ), - ) - else: - timeout = timedelta(seconds=cfg.get("timeout", 30)) - sse_read_timeout = timedelta( - seconds=cfg.get("sse_read_timeout", 60 * 5), - ) - self._streams_context = streamablehttp_client( - url=cfg["url"], - headers=cfg.get("headers", {}), - timeout=timeout, - sse_read_timeout=sse_read_timeout, - terminate_on_close=cfg.get("terminate_on_close", True), - ) - read_s, write_s, _ = await self.exit_stack.enter_async_context( - self._streams_context, - ) - - # Create a new client session - read_timeout = timedelta(seconds=cfg.get("session_read_timeout", 60)) - self.session = await self.exit_stack.enter_async_context( - mcp.ClientSession( - read_stream=read_s, - write_stream=write_s, - read_timeout_seconds=read_timeout, - logging_callback=logging_callback, # type: ignore - ), - ) - - else: - cfg = _prepare_stdio_env(cfg) - server_params = mcp.StdioServerParameters( - **cfg, - ) - - def callback(msg: str | mcp.types.LoggingMessageNotificationParams) -> None: - # Handle MCP service error logs - if isinstance(msg, mcp.types.LoggingMessageNotificationParams): - if msg.level in ( - "warning", - "error", - "critical", - "alert", - "emergency", - ): - log_msg = f"[{msg.level.upper()}] {msg.data!s}" - self.server_errlogs.append(log_msg) - - stdio_transport = await self.exit_stack.enter_async_context( - mcp.stdio_client( - server_params, - errlog=cast( - Any, - LogPipe( - level=logging.INFO, - logger=logger, - identifier=f"MCPServer-{name}", - callback=callback, - ), - ), - ), - ) - self.process_pid = self._extract_stdio_process_pid(stdio_transport) - - # Create a new client session - self.session = await self.exit_stack.enter_async_context( - mcp.ClientSession(*stdio_transport), - ) - await self.session.initialize() - - async def list_tools_and_save(self) -> mcp.ListToolsResult: - """List all tools from the server and save them to self.tools""" - if not self.session: - raise Exception("MCP Client is not initialized") - response = await self.session.list_tools() - self.tools = response.tools - return response - - async def _reconnect(self) -> None: - """Reconnect to the MCP server using the stored configuration. - - Uses asyncio.Lock to ensure thread-safe reconnection in concurrent environments. - - Raises: - Exception: raised when reconnection fails - """ - async with self._reconnect_lock: - # Check if already reconnecting (useful for logging) - if self._reconnecting: - logger.debug( - f"MCP Client {self._server_name} is already reconnecting, skipping" - ) - return - - if not self._mcp_server_config or not self._server_name: - raise Exception("Cannot reconnect: missing connection configuration") - - self._reconnecting = True - try: - logger.info( - f"Attempting to reconnect to MCP server {self._server_name}..." - ) - - # Save old exit_stack for later cleanup (don't close it now to avoid cancel scope issues) - if self.exit_stack: - self._old_exit_stacks.append(self.exit_stack) - - # Mark old session as invalid - self.session = None - - # Create new exit stack for new connection - self.exit_stack = AsyncExitStack() - - # Reconnect using stored config - await self.connect_to_server(self._mcp_server_config, self._server_name) - await self.list_tools_and_save() - - logger.info( - f"Successfully reconnected to MCP server {self._server_name}" - ) - except Exception as e: - logger.error( - f"Failed to reconnect to MCP server {self._server_name}: {e}" - ) - raise - finally: - self._reconnecting = False - - async def call_tool_with_reconnect( - self, - tool_name: str, - arguments: dict, - read_timeout_seconds: timedelta, - ) -> mcp.types.CallToolResult: - """Call MCP tool with automatic reconnection on failure, max 2 retries. - - Args: - tool_name: tool name - arguments: tool arguments - read_timeout_seconds: read timeout - - Returns: - MCP tool call result - - Raises: - ValueError: MCP session is not available - anyio.ClosedResourceError: raised after reconnection failure - """ - - @retry( - retry=retry_if_exception_type(anyio.ClosedResourceError), - stop=stop_after_attempt(2), - wait=wait_exponential(multiplier=1, min=1, max=3), - before_sleep=before_sleep_log(TenacityLogger(logger), logging.WARNING), - reraise=True, - ) - async def _call_with_retry(): - if not self.session: - raise ValueError("MCP session is not available for MCP function tools.") - - try: - return await self.session.call_tool( - name=tool_name, - arguments=arguments, - read_timeout_seconds=read_timeout_seconds, - ) - except anyio.ClosedResourceError: - logger.warning( - f"MCP tool {tool_name} call failed (ClosedResourceError), attempting to reconnect..." - ) - # Attempt to reconnect - await self._reconnect() - # Reraise the exception to trigger tenacity retry - raise - - return await _call_with_retry() - - async def cleanup(self) -> None: - """Clean up resources including old exit stacks from reconnections""" - # Close current exit stack - try: - await self.exit_stack.aclose() - except Exception as e: - logger.debug(f"Error closing current exit stack: {e}") - - # Don't close old exit stacks as they may be in different task contexts - # They will be garbage collected naturally - # Just clear the list to release references - self._old_exit_stacks.clear() - - # Set running_event first to unblock any waiting tasks - self.running_event.set() - self.process_pid = None diff --git a/astrbot/_internal/protocols/mcp/config.py b/astrbot/_internal/protocols/mcp/config.py deleted file mode 100644 index 0ea528948..000000000 --- a/astrbot/_internal/protocols/mcp/config.py +++ /dev/null @@ -1,55 +0,0 @@ -"""MCP configuration management.""" - -import json -import os - -from astrbot.core.utils.astrbot_path import get_astrbot_data_path - -DEFAULT_MCP_CONFIG = {"mcpServers": {}} - - -def get_mcp_config_path() -> str: - """Get the path to the MCP configuration file.""" - data_dir = get_astrbot_data_path() - return os.path.join(data_dir, "mcp_server.json") - - -def load_mcp_config() -> dict: - """Load MCP configuration from file. - - Returns: - MCP configuration dict. If file doesn't exist, returns default config. - - """ - config_path = get_mcp_config_path() - if not os.path.exists(config_path): - # Create default config if not exists - os.makedirs(os.path.dirname(config_path), exist_ok=True) - with open(config_path, "w", encoding="utf-8") as f: - json.dump(DEFAULT_MCP_CONFIG, f, ensure_ascii=False, indent=4) - return DEFAULT_MCP_CONFIG - - try: - with open(config_path, encoding="utf-8") as f: - return json.load(f) - except Exception: - return DEFAULT_MCP_CONFIG - - -def save_mcp_config(config: dict) -> bool: - """Save MCP configuration to file. - - Args: - config: MCP configuration dict to save. - - Returns: - True if successful, False otherwise. - - """ - config_path = get_mcp_config_path() - try: - with open(config_path, "w", encoding="utf-8") as f: - json.dump(config, f, ensure_ascii=False, indent=4) - return True - except Exception: - return False diff --git a/astrbot/_internal/protocols/mcp/tool.py b/astrbot/_internal/protocols/mcp/tool.py deleted file mode 100644 index 5e9eda1db..000000000 --- a/astrbot/_internal/protocols/mcp/tool.py +++ /dev/null @@ -1,45 +0,0 @@ -"""MCP tool wrapper.""" - -from datetime import timedelta -from typing import TYPE_CHECKING, Any - -try: - import mcp -except (ModuleNotFoundError, ImportError): - mcp = None # type: ignore - -from astrbot._internal.tools.base import FunctionTool - -if TYPE_CHECKING: - from astrbot._internal.protocols.mcp.client import McpClient - - -class MCPTool(FunctionTool): - """A function tool that calls an MCP service.""" - - def __init__( - self, - mcp_tool: "mcp.types.Tool", - mcp_client: "McpClient", - mcp_server_name: str, - **kwargs: Any, - ) -> None: - super().__init__( - name=mcp_tool.name, - description=mcp_tool.description or "", - parameters=mcp_tool.inputSchema, - ) - self.mcp_tool = mcp_tool - self.mcp_client = mcp_client - self.mcp_server_name = mcp_server_name - self.source = "mcp" - - async def call(self, **kwargs: Any) -> Any: - """Call the MCP tool with the given arguments.""" - # Note: For actual usage, context.tool_call_timeout is needed - # but for simplicity we use a default timeout here - return await self.mcp_client.call_tool_with_reconnect( - tool_name=self.mcp_tool.name, - arguments=kwargs, - read_timeout_seconds=timedelta(seconds=60), - ) diff --git a/astrbot/_internal/runtime/__init__.py b/astrbot/_internal/runtime/__init__.py deleted file mode 100644 index 38d1843cd..000000000 --- a/astrbot/_internal/runtime/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from astrbot._internal.runtime.__main__ import bootstrap - -__all__ = ["bootstrap"] diff --git a/astrbot/_internal/runtime/__main__.py b/astrbot/_internal/runtime/__main__.py deleted file mode 100644 index 120195161..000000000 --- a/astrbot/_internal/runtime/__main__.py +++ /dev/null @@ -1,24 +0,0 @@ -from __future__ import annotations - -import anyio - -from astrbot._internal.abc.base_astrbot_gateway import BaseAstrbotGateway -from astrbot._internal.abc.base_astrbot_orchestrator import BaseAstrbotOrchestrator -from astrbot._internal.geteway.server import AstrbotGateway -from astrbot._internal.runtime.orchestrator import AstrbotOrchestrator - - -async def bootstrap(): - orchestrator: BaseAstrbotOrchestrator = AstrbotOrchestrator() - gw: BaseAstrbotGateway = AstrbotGateway(orchestrator) - - # anyio 的结构化并发 - async with anyio.create_task_group() as tg: - tg.start_soon(orchestrator.lsp.connect) # 启动 LSP client - tg.start_soon(orchestrator.mcp.connect) # 启动 MCP client - tg.start_soon(orchestrator.acp.connect) # 启动 ACP client - tg.start_soon(orchestrator.abp.connect) # 启动 ABP client - await anyio.sleep(0.5) - tg.start_soon(orchestrator.run_loop) # 启动编排器循环 - - tg.start_soon(gw.serve) # 面板后端服务 diff --git a/astrbot/_internal/runtime/orchestrator.py b/astrbot/_internal/runtime/orchestrator.py deleted file mode 100644 index 8211fe0c4..000000000 --- a/astrbot/_internal/runtime/orchestrator.py +++ /dev/null @@ -1,164 +0,0 @@ -""" -AstrBot Orchestrator - core runtime that coordinates all protocol clients. - -The orchestrator manages the lifecycle of LSP, MCP, ACP, and ABP clients, -and runs the main event loop that dispatches messages between components. -""" - -from __future__ import annotations - -from typing import Any - -import anyio - -from astrbot import logger -from astrbot._internal.abc.base_astrbot_orchestrator import BaseAstrbotOrchestrator -from astrbot._internal.protocols.abp.client import AstrbotAbpClient -from astrbot._internal.protocols.acp.client import AstrbotAcpClient -from astrbot._internal.protocols.lsp.client import AstrbotLspClient -from astrbot._internal.protocols.mcp.client import McpClient -from astrbot._internal.stars import RuntimeStatusStar - -log = logger - - -class AstrbotOrchestrator(BaseAstrbotOrchestrator): - """ - Core runtime orchestrator for AstrBot. - - Manages: - - LSP client: Language Server Protocol for editor integrations - - MCP client: Model Context Protocol for external tool servers - - ACP client: AstrBot Communication Protocol for inter-service communication - - ABP client: AstrBot Protocol for built-in star (plugin) communication - """ - - def __init__(self) -> None: - # Initialize protocol clients (use concrete types for full method access) - self.lsp = AstrbotLspClient() - self.mcp = McpClient() - self.acp = AstrbotAcpClient() - self.abp = AstrbotAbpClient() - - self._running = False - self._stars: dict[str, Any] = {} - self._message_count: int = 0 - self._last_activity_timestamp: float | None = None - - # Auto-register RuntimeStatusStar - self._runtime_status_star = RuntimeStatusStar() - self._runtime_status_star.set_orchestrator(self) - self._stars["runtime-status-star"] = self._runtime_status_star - self.abp.register_star("runtime-status-star", self._runtime_status_star) - - log.debug("AstrbotOrchestrator initialized.") - - async def start(self) -> None: - """ - Initialize all protocol clients. - - Calls connect() on all protocol clients to prepare them for use. - """ - log.info("Starting AstrbotOrchestrator...") - - await self.lsp.connect() - await self.mcp.connect() - await self.acp.connect() - await self.abp.connect() - - self._running = True - log.info("AstrbotOrchestrator started.") - - async def run_loop(self) -> None: - """ - Main orchestrator event loop. - - This loop runs continuously, handling: - - Periodic health checks of protocol clients - - Message routing between protocols - - Star (plugin) lifecycle management - """ - self._running = True - log.info("AstrbotOrchestrator run loop started.") - - stop_event = anyio.Event() - - def set_stop() -> None: - stop_event.set() - - # Store the callback for cleanup - self._stop_callback = set_stop - - try: - while self._running: - # TODO: Periodic tasks: - # - Check LSP server health - # - Check MCP session status - # - Check ACP client connections - # - Process any pending star notifications - - # Wait for 5 seconds or until shutdown is called - with anyio.move_on_after(5): - await stop_event.wait() - - except anyio.get_cancelled_exc_class(): - log.info("Orchestrator run loop cancelled.") - finally: - self._running = False - self._stop_callback = None - log.info("AstrbotOrchestrator run loop stopped.") - - async def register_star(self, name: str, star_instance: Any) -> None: - """ - Register a star (plugin) with the orchestrator. - - Args: - name: Unique name for the star - star_instance: Star plugin instance - """ - self._stars[name] = star_instance - self.abp.register_star(name, star_instance) - log.info(f"Star '{name}' registered.") - - async def unregister_star(self, name: str) -> None: - """ - Unregister a star (plugin) from the orchestrator. - - Args: - name: Name of the star to unregister - """ - self._stars.pop(name, None) - self.abp.unregister_star(name) - log.info(f"Star '{name}' unregistered.") - - async def get_star(self, name: str) -> Any | None: - """Get a registered star by name.""" - return self._stars.get(name) - - async def list_stars(self) -> list[str]: - """List all registered star names.""" - return list(self._stars.keys()) - - def record_activity(self) -> None: - """Record a message activity for stats tracking.""" - self._message_count += 1 - import time - - self._last_activity_timestamp = time.time() - - async def shutdown(self) -> None: - """ - Shutdown the orchestrator and all protocol clients. - """ - log.info("Shutting down AstrbotOrchestrator...") - self._running = False - - # Shutdown all protocol clients - await self.lsp.shutdown() - await self.acp.shutdown() - await self.abp.shutdown() - - # MCP cleanup - await self.mcp.cleanup() - - log.info("AstrbotOrchestrator shut down.") diff --git a/astrbot/_internal/runtime/rust/__init__.py b/astrbot/_internal/runtime/rust/__init__.py deleted file mode 100644 index de05c36ec..000000000 --- a/astrbot/_internal/runtime/rust/__init__.py +++ /dev/null @@ -1,18 +0,0 @@ -import sys - -try: - from ._core import cli as _cli - - def cli(): - if len(sys.argv) == 1: - sys.argv.append("--help") - return _cli() -except ImportError: - from click import echo - - def cli(): - echo(""" - AstrBot CLI(rust) is not available. - Developer: maturin dev - User: uv run astrbot-rs - """) diff --git a/astrbot/_internal/runtime/rust/_core.pyi b/astrbot/_internal/runtime/rust/_core.pyi deleted file mode 100644 index 41d6f8492..000000000 --- a/astrbot/_internal/runtime/rust/_core.pyi +++ /dev/null @@ -1,16 +0,0 @@ -from typing import Any - -class AstrbotOrchestrator: - def start(self) -> None: ... - def stop(self) -> None: ... - def is_running(self) -> bool: ... - def register_star(self, name: str, handler: str) -> None: ... - def unregister_star(self, name: str) -> None: ... - def list_stars(self) -> list[str]: ... - def record_activity(self) -> None: ... - def get_stats(self) -> dict[str, Any]: ... - def set_protocol_connected(self, protocol: str, connected: bool) -> None: ... - def get_protocol_status(self, protocol: str) -> dict[str, Any] | None: ... - -def get_orchestrator() -> AstrbotOrchestrator: ... -def cli() -> None: ... diff --git a/astrbot/_internal/skills/__init__.py b/astrbot/_internal/skills/__init__.py deleted file mode 100644 index e36af0aed..000000000 --- a/astrbot/_internal/skills/__init__.py +++ /dev/null @@ -1,13 +0,0 @@ -"""Internal skills module - re-exports from core.skills.skill_manager.""" - -from astrbot.core.skills.skill_manager import ( - SkillInfo, - SkillManager, - build_skills_prompt, -) - -__all__ = [ - "SkillInfo", - "SkillManager", - "build_skills_prompt", -] diff --git a/astrbot/_internal/stars/__init__.py b/astrbot/_internal/stars/__init__.py deleted file mode 100644 index 2e44bc8db..000000000 --- a/astrbot/_internal/stars/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -""" -Stars (built-in plugins) for AstrBot runtime. -""" - -from astrbot._internal.stars.runtime_status_star import RuntimeStatusStar - -__all__ = ["RuntimeStatusStar"] diff --git a/astrbot/_internal/stars/runtime_status_star.py b/astrbot/_internal/stars/runtime_status_star.py deleted file mode 100644 index eaa396ed4..000000000 --- a/astrbot/_internal/stars/runtime_status_star.py +++ /dev/null @@ -1,127 +0,0 @@ -""" -RuntimeStatusStar - ABP plugin that exposes core runtime internal state. - -This star provides tools for querying: -- Runtime status (running state, uptime) -- Protocol client status (LSP, MCP, ACP, ABP) -- Registered stars registry -- Message counts and metrics -""" - -from __future__ import annotations - -import time -from dataclasses import dataclass, field -from datetime import datetime, timezone -from typing import Any - - -@dataclass -class RuntimeStatusStar: - """ - ABP star that exposes core runtime internal state as callable tools. - - Tools provided: - - get_runtime_status: Returns running state and uptime - - get_protocol_status: Returns LSP, MCP, ACP, ABP status - - get_star_registry: Returns registered star names - - get_stats: Returns message counts and metrics - """ - - name: str = "runtime-status-star" - description: str = "ABP plugin that exposes core runtime internal state" - - _start_time: float = field(default_factory=time.time, init=False) - _orchestrator: Any = field(default=None, init=False) - - def set_orchestrator(self, orchestrator: Any) -> None: - """Set the orchestrator reference for status queries.""" - self._orchestrator = orchestrator - - async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: - """ - Handle tool calls from ABP client. - - Args: - tool_name: Name of the tool to call - arguments: Tool arguments - - Returns: - Tool result - """ - if tool_name == "get_runtime_status": - return self._get_runtime_status() - elif tool_name == "get_protocol_status": - return await self._get_protocol_status() - elif tool_name == "get_star_registry": - return await self._get_star_registry() - elif tool_name == "get_stats": - return self._get_stats() - else: - raise ValueError(f"Unknown tool: {tool_name}") - - def _get_runtime_status(self) -> dict[str, Any]: - """Get overall runtime state.""" - running = ( - getattr(self._orchestrator, "running", False) - if self._orchestrator - else False - ) - uptime_seconds = time.time() - self._start_time - return { - "running": running, - "uptime_seconds": uptime_seconds, - } - - async def _get_protocol_status(self) -> dict[str, Any]: - """Get status of each protocol client.""" - if not self._orchestrator: - return { - "lsp": {"connected": False, "name": "lsp-client"}, - "mcp": {"connected": False, "name": "mcp-client"}, - "acp": {"connected": False, "name": "acp-client"}, - "abp": {"connected": False, "name": "abp-client"}, - } - - return { - "lsp": { - "connected": getattr(self._orchestrator.lsp, "connected", False), - "name": "lsp-client", - }, - "mcp": { - "connected": getattr(self._orchestrator.mcp, "connected", False), - "name": "mcp-client", - }, - "acp": { - "connected": getattr(self._orchestrator.acp, "connected", False), - "name": "acp-client", - }, - "abp": { - "connected": getattr(self._orchestrator.abp, "connected", False), - "name": "abp-client", - }, - } - - async def _get_star_registry(self) -> dict[str, Any]: - """Get list of registered stars.""" - if not self._orchestrator: - return {"stars": []} - - stars = await self._orchestrator.list_stars() - return {"stars": stars} - - def _get_stats(self) -> dict[str, Any]: - """Get message counts and metrics.""" - result: dict[str, Any] = { - "uptime_seconds": time.time() - self._start_time, - } - if self._orchestrator: - result["total_messages"] = getattr(self._orchestrator, "_message_count", 0) - last_ts = getattr(self._orchestrator, "_last_activity_timestamp", None) - if last_ts is not None: - result["last_activity"] = datetime.fromtimestamp( - last_ts, tz=timezone.utc - ).isoformat() - else: - result["last_activity"] = None - return result diff --git a/astrbot/_internal/tools/__init__.py b/astrbot/_internal/tools/__init__.py deleted file mode 100644 index 434182911..000000000 --- a/astrbot/_internal/tools/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -"""Internal tools module for AstrBot runtime.""" - -from .base import FunctionTool, ToolSet - -__all__ = ["FunctionTool", "ToolSet"] diff --git a/astrbot/_internal/tools/base.py b/astrbot/_internal/tools/base.py deleted file mode 100644 index aaa9c2ad3..000000000 --- a/astrbot/_internal/tools/base.py +++ /dev/null @@ -1,332 +0,0 @@ -"""Base tool classes for AstrBot internal runtime. - -This module provides the FunctionTool base class used by MCP tools -in the new internal architecture. -""" - -import copy -from collections.abc import AsyncGenerator, Awaitable, Callable, Iterator -from dataclasses import dataclass, field -from typing import Any - -from pydantic import model_validator - -ParametersType = dict[str, Any] - - -@dataclass -class ToolSchema: - """A class representing the schema of a tool for function calling.""" - - name: str - """The name of the tool.""" - - description: str - """The description of the tool.""" - - parameters: ParametersType = field(default_factory=dict) - """The parameters of the tool, in JSON Schema format.""" - - @model_validator(mode="after") - def validate_parameters(self) -> "ToolSchema": - """Validate the parameters JSON schema.""" - import jsonschema - - jsonschema.validate( - self.parameters, jsonschema.Draft202012Validator.META_SCHEMA - ) - return self - - -@dataclass -class FunctionTool(ToolSchema): - """A callable tool, for function calling.""" - - handler: Callable[..., Awaitable[str | None] | AsyncGenerator[Any, None]] | None = ( - None - ) - """a callable that implements the tool's functionality. It should be an async function.""" - - handler_module_path: str | None = None - """ - The module path of the handler function. This is empty when the origin is mcp. - This field must be retained, as the handler will be wrapped in functools.partial during initialization, - causing the handler's __module__ to be functools - """ - - active: bool = True - """ - Whether the tool is active. This field is a special field for AstrBot. - You can ignore it when integrating with other frameworks. - """ - - is_background_task: bool = False - """ - Declare this tool as a background task. Background tasks return immediately - with a task identifier while the real work continues asynchronously. - """ - - source: str = "mcp" - """ - Origin of this tool: 'plugin' (from star plugins), 'internal' (AstrBot built-in), - or 'mcp' (from MCP servers). Used by WebUI for display grouping. - """ - - def __repr__(self) -> str: - return f"FuncTool(name={self.name}, parameters={self.parameters}, description={self.description})" - - async def call(self, **kwargs: Any) -> Any: - """Run the tool with the given arguments. The handler field has priority.""" - raise NotImplementedError( - "FunctionTool.call() must be implemented by subclasses or set a handler." - ) - - -class ToolSet: - """ - A collection of FunctionTools grouped under a namespace. - - ToolSets allow organizing related tools together. The LLM sees tools - as "namespace/tool_name" when calling. - """ - - def __init__(self, namespace: str, tools: list[FunctionTool] | None = None) -> None: - self.namespace = namespace - self._tools: dict[str, FunctionTool] = {} - if tools: - for tool in tools: - self.add(tool) - - def add(self, tool: FunctionTool) -> None: - """Add a tool to the set.""" - self._tools[tool.name] = tool - - def add_tool(self, tool: FunctionTool) -> None: - """Add a tool to the set (alias for add()).""" - self.add(tool) - - def remove(self, name: str) -> FunctionTool | None: - """Remove and return a tool by name.""" - return self._tools.pop(name, None) - - def remove_tool(self, name: str) -> None: - """Remove a tool by its name.""" - self._tools.pop(name, None) - - def get(self, name: str) -> FunctionTool | None: - """Get a tool by name.""" - return self._tools.get(name) - - def get_tool(self, name: str) -> FunctionTool | None: - """Get a tool by name (alias for get).""" - return self.get(name) - - def list_tools(self) -> list[FunctionTool]: - """List all tools in this set.""" - return list(self._tools.values()) - - def __iter__(self) -> Iterator[FunctionTool]: - return iter(self._tools.values()) - - def __len__(self) -> int: - return len(self._tools) - - def __bool__(self) -> bool: - return bool(self._tools) - - def __repr__(self) -> str: - return f"ToolSet(namespace={self.namespace!r}, tools={self.list_tools()!r})" - - def __str__(self) -> str: - return f"ToolSet({self.namespace}, {len(self)} tools)" - - def names(self) -> list[str]: - """Get names of all tools in this set.""" - return [tool.name for tool in self.tools] - - def empty(self) -> bool: - """Check if the tool set is empty.""" - return len(self) == 0 - - def merge(self, other: "ToolSet") -> None: - """Merge another ToolSet into this one.""" - for tool in other.tools: - self.add(tool) - - def normalize(self) -> None: - """Sort tools by name for deterministic serialization.""" - self._tools = dict(sorted(self._tools.items(), key=lambda x: x[0])) - - def get_light_tool_set(self) -> "ToolSet": - """Return a light tool set with only name/description.""" - light_tools = [] - for tool in self.tools: - if hasattr(tool, "active") and not tool.active: - continue - light_tools.append( - FunctionTool( - name=tool.name, - description=tool.description, - parameters={"type": "object", "properties": {}}, - handler=None, - ) - ) - return ToolSet("default", light_tools) - - def get_param_only_tool_set(self) -> "ToolSet": - """Return a tool set with name/parameters only (no description).""" - param_tools = [] - for tool in self.tools: - if hasattr(tool, "active") and not tool.active: - continue - params = ( - copy.deepcopy(tool.parameters) - if tool.parameters - else {"type": "object", "properties": {}} - ) - param_tools.append( - FunctionTool( - name=tool.name, - description="", - parameters=params, - handler=None, - ) - ) - return ToolSet("default", param_tools) - - @property - def tools(self) -> list[FunctionTool]: - """List all tools in this set.""" - return list(self._tools.values()) - - def openai_schema( - self, omit_empty_parameter_field: bool = False - ) -> list[dict[str, Any]]: - """Convert tools to OpenAI API function calling schema format.""" - result: list[dict[str, Any]] = [] - for tool in self._tools.values(): - func_def: dict[str, Any] = { - "type": "function", - "function": {"name": tool.name}, - } - if tool.description: - func_def["function"]["description"] = tool.description - - if tool.parameters is not None: - if ( - tool.parameters.get("properties") - ) or not omit_empty_parameter_field: - func_def["function"]["parameters"] = tool.parameters - - result.append(func_def) - return result - - def anthropic_schema(self) -> list[dict]: - """Convert tools to Anthropic API format.""" - result = [] - for tool in self.tools: - input_schema: dict[str, Any] = {"type": "object"} - if tool.parameters: - input_schema["properties"] = tool.parameters.get("properties", {}) - input_schema["required"] = tool.parameters.get("required", []) - tool_def: dict[str, Any] = {"name": tool.name, "input_schema": input_schema} - if tool.description: - tool_def["description"] = tool.description - result.append(tool_def) - return result - - def google_schema(self) -> dict: - """Convert tools to Google GenAI API format.""" - - def convert_schema(schema: dict) -> dict: - supported_types = { - "string", - "number", - "integer", - "boolean", - "array", - "object", - "null", - } - supported_formats = { - "string": {"enum", "date-time"}, - "integer": {"int32", "int64"}, - "number": {"float", "double"}, - } - - if "anyOf" in schema: - return {"anyOf": [convert_schema(s) for s in schema["anyOf"]]} - - result = {} - origin_type = schema.get("type") - target_type = origin_type - - if isinstance(origin_type, list): - target_type = next((t for t in origin_type if t != "null"), "string") - - if target_type in supported_types: - result["type"] = target_type - if "format" in schema and schema["format"] in supported_formats.get(result["type"], set()): - result["format"] = schema["format"] - else: - result["type"] = "null" - - support_fields = { - "title", - "description", - "enum", - "minimum", - "maximum", - "maxItems", - "minItems", - "nullable", - "required", - } - result.update({k: schema[k] for k in support_fields if k in schema}) - - if "properties" in schema: - properties = {} - for key, value in schema["properties"].items(): - prop_value = convert_schema(value) - if "default" in prop_value: - del prop_value["default"] - if "additionalProperties" in prop_value: - del prop_value["additionalProperties"] - properties[key] = prop_value - if properties: - result["properties"] = properties - - if target_type == "array": - items_schema = schema.get("items") - if isinstance(items_schema, dict): - result["items"] = convert_schema(items_schema) - else: - result["items"] = {"type": "string"} - - return result - - tools_list = [] - for tool in self.tools: - d: dict[str, Any] = {"name": tool.name} - if tool.description: - d["description"] = tool.description - if tool.parameters: - d["parameters"] = convert_schema(tool.parameters) - tools_list.append(d) - - declarations: dict[str, Any] = {} - if tools_list: - declarations["function_declarations"] = tools_list - return declarations - - def get_func_desc_openai_style(self, omit_empty_parameter_field: bool = False): - """Get tools in OpenAI function calling style (deprecated).""" - return self.openai_schema(omit_empty_parameter_field) - - def get_func_desc_anthropic_style(self): - """Get tools in Anthropic style (deprecated).""" - return self.anthropic_schema() - - def get_func_desc_google_genai_style(self): - """Get tools in Google GenAI style (deprecated).""" - return self.google_schema() diff --git a/astrbot/_internal/tools/builtin.py b/astrbot/_internal/tools/builtin.py deleted file mode 100644 index c2d823a9a..000000000 --- a/astrbot/_internal/tools/builtin.py +++ /dev/null @@ -1,48 +0,0 @@ -""" -Builtin tools for AstrBot - re-exports from core.tools for backward compatibility. - -This module re-exports the builtin tools (cron, send_message, kb_query) from -the deprecated core.tools module for backward compatibility. - -TODO: These tools should be fully migrated to _internal and core.tools -should be removed once all consumers update their imports. -""" - -from __future__ import annotations - -# Re-export cron tools -from astrbot.core.tools.cron_tools import ( - CREATE_CRON_JOB_TOOL, - DELETE_CRON_JOB_TOOL, - LIST_CRON_JOBS_TOOL, - CreateActiveCronTool, - DeleteCronJobTool, - ListCronJobsTool, -) - -# Re-export knowledge_base_query tool -from astrbot.core.tools.kb_query import ( - KNOWLEDGE_BASE_QUERY_TOOL, - KnowledgeBaseQueryTool, -) - -# Re-export send_message tool -from astrbot.core.tools.send_message import ( - SEND_MESSAGE_TO_USER_TOOL, - SendMessageToUserTool, -) - -__all__ = [ - # Cron tools - "CREATE_CRON_JOB_TOOL", - "DELETE_CRON_JOB_TOOL", - "KNOWLEDGE_BASE_QUERY_TOOL", - "LIST_CRON_JOBS_TOOL", - "SEND_MESSAGE_TO_USER_TOOL", - # Classes - "CreateActiveCronTool", - "DeleteCronJobTool", - "KnowledgeBaseQueryTool", - "ListCronJobsTool", - "SendMessageToUserTool", -] diff --git a/astrbot/_internal/tools/registry.py b/astrbot/_internal/tools/registry.py deleted file mode 100644 index 314f25739..000000000 --- a/astrbot/_internal/tools/registry.py +++ /dev/null @@ -1,278 +0,0 @@ -"""Tools registry for AstrBot internal runtime.""" - -from __future__ import annotations - -from typing import Any - -# Re-export from base -from astrbot._internal.tools.base import FunctionTool, ToolSet - -__all__ = [ - "DEFAULT_MCP_CONFIG", - "ENABLE_MCP_TIMEOUT_ENV", - "FuncCall", - "FunctionTool", - "FunctionToolManager", - "MCPAllServicesFailedError", - "MCPInitError", - "MCPInitSummary", - "MCPInitTimeoutError", - "MCPShutdownTimeoutError", - "ToolSet", -] - - -# MCP config constants (re-exported from protocols) -try: - from astrbot._internal.protocols.mcp import ( - DEFAULT_MCP_CONFIG, - MCPAllServicesFailedError, - MCPInitError, - MCPInitSummary, - MCPInitTimeoutError, - MCPShutdownTimeoutError, - ) -except ImportError: - DEFAULT_MCP_CONFIG: dict[str, Any] = {} - MCPAllServicesFailedError: type[Exception] = Exception - MCPInitError: type[Exception] = Exception - MCPInitSummary: type[dict] = dict - MCPInitTimeoutError: type[TimeoutError] = TimeoutError - MCPShutdownTimeoutError: type[TimeoutError] = TimeoutError - -ENABLE_MCP_TIMEOUT_ENV = "ASTRBOT_MCP_TIMEOUT_ENABLED" -MCP_INIT_TIMEOUT_ENV = "ASTRBOT_MCP_INIT_TIMEOUT" - - -class FunctionToolManager: - """Central registry for all function tools.""" - - def __init__(self) -> None: - self._func_list: list[FunctionTool] = [] - - @property - def func_list(self) -> list[FunctionTool]: - """Get the list of function tools.""" - return self._func_list - - @func_list.setter - def func_list(self, value: list[FunctionTool]) -> None: - """Set the list of function tools.""" - self._func_list = value - - def add(self, tool: FunctionTool) -> None: - """Add a tool to the registry.""" - self._func_list.append(tool) - - def remove(self, name: str) -> bool: - """Remove a tool by name. Returns True if found.""" - for i, f in enumerate(self._func_list): - if f.name == name: - self._func_list.pop(i) - return True - return False - - def get_func(self, name: str) -> FunctionTool | None: - """Get a tool by name. Returns the last active tool if multiple match.""" - last_match: FunctionTool | None = None - for f in reversed(self._func_list): - if f.name == name: - if getattr(f, "active", True): - return f - if last_match is None: - last_match = f - return last_match - - def get_full_tool_set(self) -> ToolSet: - """Return a ToolSet with all active tools, deduplicated by name.""" - seen: dict[str, FunctionTool] = {} - for tool in reversed(self._func_list): - if tool.name not in seen and getattr(tool, "active", True): - seen[tool.name] = tool - return ToolSet("default", list(seen.values())) - - def register_internal_tools(self) -> None: - """Register built-in computer tools (shell, python, browser, neo).""" - # Import here to avoid circular imports - from astrbot.core.computer.computer_tool_provider import get_all_tools - - for tool in get_all_tools(): - if self.get_func(tool.name) is None: - self.add(tool) - - # MCP-related stub methods for base class compatibility - async def enable_mcp_server( - self, name: str, config: dict[str, Any], init_timeout: int = 30 - ) -> None: - """Enable an MCP server (stub).""" - pass - - async def disable_mcp_server( - self, name: str = "", timeout: int = 10, shutdown_timeout: int = 10 - ) -> None: - """Disable an MCP server (stub).""" - pass - - async def init_mcp_clients(self) -> None: - """Initialize MCP clients (stub).""" - pass - - async def test_mcp_server_connection( - self, config: dict[str, Any] - ) -> tuple[bool, str]: - """Test MCP server connection (stub).""" - return False, "Not implemented" - - async def sync_modelscope_mcp_servers(self) -> None: - """Sync ModelScope MCP servers (stub).""" - pass - - def load_mcp_config(self) -> dict[str, Any]: - """Load MCP configuration (stub).""" - return {"mcpServers": {}} - - def save_mcp_config(self, config: dict[str, Any]) -> bool: - """Save MCP configuration (stub).""" - return True - - def activate_llm_tool(self, name: str) -> bool: - """Activate an LLM tool (stub).""" - return True - - def deactivate_llm_tool(self, name: str) -> bool: - """Deactivate an LLM tool (stub).""" - return True - - @property - def mcp_client_dict(self) -> dict[str, Any]: - """Return dict of MCP clients (stub).""" - return {} - - @property - def mcp_server_runtime_view(self) -> dict[str, Any]: - """Return runtime view of MCP servers (stub).""" - return {} - - -class FuncCall(FunctionToolManager): - """Alias for FunctionToolManager for backward compatibility.""" - - def __init__(self) -> None: - super().__init__() - self._mcp_server_runtime_view: dict[str, Any] = {} - self._mcp_client_dict: dict[str, Any] = {} - - @property - def mcp_server_runtime_view(self) -> dict[str, Any]: - """Return runtime view of MCP servers.""" - return self._mcp_server_runtime_view - - @property - def mcp_client_dict(self) -> dict[str, Any]: - """Return dict of MCP clients (for backward compatibility).""" - return self._mcp_client_dict - - async def init_mcp_clients(self) -> None: - """Initialize MCP clients (stub implementation).""" - pass - - def add_func( - self, - name: str, - func_args: list[dict[str, Any]], - desc: str, - handler: Any, - ) -> None: - """Add a function tool (deprecated, use add() instead).""" - params: dict[str, Any] = { - "type": "object", - "properties": {}, - } - for param in func_args: - params["properties"][param["name"]] = { - "type": param.get("type", "string"), - "description": param.get("description", ""), - } - func = FunctionTool( - name=name, - parameters=params, - description=desc, - handler=handler, - ) - self.add(func) - - def remove_func(self, name: str) -> None: - """Remove a function tool by name (deprecated, use remove() instead).""" - self.remove(name) - - def get_func(self, name: str) -> FunctionTool | None: - """Get a function tool by name.""" - return super().get_func(name) - - def names(self) -> list[str]: - """Get all tool names.""" - return [f.name for f in self.func_list] - - def remove_tool(self, name: str) -> None: - """Remove a tool by its name (alias for remove).""" - self.remove(name) - - def get_func_desc_openai_style( - self, omit_empty_parameter_field: bool = False - ) -> list[dict[str, Any]]: - """Get tools in OpenAI style (deprecated, use get_full_tool_set().openai_schema()).""" - tool_set = self.get_full_tool_set() - return tool_set.openai_schema(omit_empty_parameter_field) - - async def enable_mcp_server( - self, name: str, config: dict[str, Any], init_timeout: int = 30 - ) -> None: - """Enable an MCP server (stub implementation).""" - pass - - async def disable_mcp_server( - self, name: str = "", timeout: int = 10, shutdown_timeout: int = 10 - ) -> None: - """Disable an MCP server (stub implementation).""" - pass - - def load_mcp_config(self) -> dict[str, Any]: - """Load MCP configuration (stub implementation).""" - return {"mcpServers": {}} - - def save_mcp_config(self, config: dict[str, Any]) -> bool: - """Save MCP configuration (stub implementation).""" - return True - - def activate_llm_tool(self, name: str) -> bool: - """Activate an LLM tool (stub implementation).""" - return True - - def deactivate_llm_tool(self, name: str) -> bool: - """Deactivate an LLM tool (stub implementation).""" - return True - - async def test_mcp_server_connection( - self, config: dict[str, Any] - ) -> tuple[bool, str]: - """Test MCP server connection (stub implementation).""" - # Import the actual test function if available - try: - from astrbot._internal.protocols.mcp.client import ( - _quick_test_mcp_connection, - ) - - success, message = await _quick_test_mcp_connection(config) - if not success: - raise Exception(message) - return success, message - except Exception as e: - raise Exception(f"MCP connection test failed: {e!s}") from e - - async def sync_modelscope_mcp_servers(self) -> None: - """Sync ModelScope MCP servers (stub implementation).""" - pass - - def get_full_tool_set(self) -> ToolSet: - """Return a ToolSet with all active tools.""" - return ToolSet("default", [t for t in self.func_list if t.active])