File size: 4,763 Bytes
5dc5419 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """MCPTool - MCP protocol tool integration for Stack 2.9"""
import json
from pathlib import Path
from typing import Any, Dict, Optional
from .base import BaseTool, ToolResult
from .registry import tool_registry
MCP_CONFIG_FILE = Path.home() / ".stack-2.9" / "mcp_config.json"
def _load_mcp_config() -> Dict[str, Any]:
"""Load MCP configuration."""
MCP_CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
if MCP_CONFIG_FILE.exists():
return json.loads(MCP_CONFIG_FILE.read_text())
return {"servers": {}}
class MCPTool(BaseTool):
"""Call an MCP server tool."""
name = "mcp_call"
description = "Call a tool on an MCP server"
input_schema = {
"type": "object",
"properties": {
"server_name": {"type": "string", "description": "MCP server name"},
"tool_name": {"type": "string", "description": "Tool to call on server"},
"arguments": {"type": "object", "description": "Arguments for the tool"}
},
"required": ["server_name", "tool_name"]
}
async def execute(self, server_name: str, tool_name: str, arguments: Optional[Dict] = None) -> ToolResult:
"""Call MCP tool."""
config = _load_mcp_config()
if server_name not in config.get("servers", {}):
return ToolResult(success=False, error=f"MCP server '{server_name}' not configured")
server = config["servers"][server_name]
return ToolResult(success=True, data={
"server": server_name,
"tool": tool_name,
"arguments": arguments or {},
"status": "simulated",
"note": f"MCP call to {server_name}/{tool_name} - requires MCP runtime"
})
class MCPServerListTool(BaseTool):
"""List configured MCP servers."""
name = "mcp_list_servers"
description = "List all configured MCP servers"
input_schema = {
"type": "object",
"properties": {},
"required": []
}
async def execute(self) -> ToolResult:
"""List MCP servers."""
config = _load_mcp_config()
servers = config.get("servers", {})
return ToolResult(success=True, data={
"servers": list(servers.keys()),
"count": len(servers)
})
class MCPServerAddTool(BaseTool):
"""Add an MCP server configuration."""
name = "mcp_add_server"
description = "Add an MCP server configuration"
input_schema = {
"type": "object",
"properties": {
"server_name": {"type": "string", "description": "Server name"},
"command": {"type": "string", "description": "Command to start server"},
"args": {"type": "array", "items": {"type": "string"}, "description": "Command arguments"},
"env": {"type": "object", "description": "Environment variables"}
},
"required": ["server_name", "command"]
}
async def execute(self, server_name: str, command: str, args: Optional[list] = None, env: Optional[Dict] = None) -> ToolResult:
"""Add MCP server."""
config = _load_mcp_config()
if "servers" not in config:
config["servers"] = {}
config["servers"][server_name] = {
"command": command,
"args": args or [],
"env": env or {},
"enabled": True
}
MCP_CONFIG_FILE.write_text(json.dumps(config, indent=2))
return ToolResult(success=True, data={
"server": server_name,
"status": "added"
})
class ReadMcpResourceTool(BaseTool):
"""Read a resource from an MCP server."""
name = "read_mcp_resource"
description = "Read a resource from an MCP server"
input_schema = {
"type": "object",
"properties": {
"server_name": {"type": "string", "description": "MCP server name"},
"resource_uri": {"type": "string", "description": "Resource URI"}
},
"required": ["server_name", "resource_uri"]
}
async def execute(self, server_name: str, resource_uri: str) -> ToolResult:
"""Read MCP resource."""
config = _load_mcp_config()
if server_name not in config.get("servers", {}):
return ToolResult(success=False, error=f"MCP server '{server_name}' not configured")
return ToolResult(success=True, data={
"server": server_name,
"resource_uri": resource_uri,
"status": "simulated",
"note": f"Read resource {resource_uri} from {server_name}"
})
# Register tools
tool_registry.register(MCPTool())
tool_registry.register(MCPServerListTool())
tool_registry.register(MCPServerAddTool())
tool_registry.register(ReadMcpResourceTool())
|