mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-27 04:20:09 +00:00
zooqueenandHanzo Dev
b144973bfc
Fix 35 ruff errors (E722, F841, B904, B905, B007, E741, S202, S310, S306, S110, TID251, UP042, F401) and run black on 526 files. Co-authored-by: Hanzo Dev <[email protected]>
141 lines
3.7 KiB
Python
141 lines
3.7 KiB
Python
"""Remove MCP servers."""
|
|
|
|
from typing import Unpack, Annotated, TypedDict, final, override
|
|
|
|
from pydantic import Field
|
|
from mcp.server.fastmcp import Context as MCPContext
|
|
|
|
from hanzo_tools.core import BaseTool, ToolContext, auto_timeout, create_tool_context
|
|
|
|
from .mcp_add import McpAddTool
|
|
|
|
ServerName = Annotated[
|
|
str,
|
|
Field(
|
|
description="Name of the server to remove",
|
|
min_length=1,
|
|
),
|
|
]
|
|
|
|
Force = Annotated[
|
|
bool,
|
|
Field(
|
|
description="Force removal even if server is running",
|
|
default=False,
|
|
),
|
|
]
|
|
|
|
|
|
class McpRemoveParams(TypedDict, total=False):
|
|
"""Parameters for MCP remove tool."""
|
|
|
|
name: str
|
|
force: bool
|
|
|
|
|
|
@final
|
|
class McpRemoveTool(BaseTool):
|
|
"""Tool for removing MCP servers."""
|
|
|
|
def __init__(self):
|
|
"""Initialize the MCP remove tool."""
|
|
pass
|
|
|
|
@property
|
|
@override
|
|
def name(self) -> str:
|
|
"""Get the tool name."""
|
|
return "mcp_remove"
|
|
|
|
@property
|
|
@override
|
|
def description(self) -> str:
|
|
"""Get the tool description."""
|
|
return """Remove previously added MCP servers.
|
|
|
|
This removes MCP servers that were added with mcp_add.
|
|
If the server is running, it will be stopped first.
|
|
|
|
Examples:
|
|
- mcp_remove --name filesystem
|
|
- mcp_remove --name github --force
|
|
|
|
Use 'mcp_stats' to see all servers before removing.
|
|
"""
|
|
|
|
@override
|
|
@auto_timeout("mcp_remove")
|
|
async def call(
|
|
self,
|
|
ctx: MCPContext,
|
|
**params: Unpack[McpRemoveParams],
|
|
) -> str:
|
|
"""Remove an MCP server.
|
|
|
|
Args:
|
|
ctx: MCP context
|
|
**params: Tool parameters
|
|
|
|
Returns:
|
|
Result of removing the server
|
|
"""
|
|
tool_ctx = create_tool_context(ctx)
|
|
await tool_ctx.set_tool_info(self.name)
|
|
|
|
# Extract parameters
|
|
name = params.get("name")
|
|
if not name:
|
|
return "Error: name is required"
|
|
|
|
force = params.get("force", False)
|
|
|
|
# Get current servers
|
|
servers = McpAddTool.get_servers()
|
|
|
|
if name not in servers:
|
|
return f"Error: Server '{name}' not found. Use 'mcp_stats' to see available servers."
|
|
|
|
server = servers[name]
|
|
|
|
await tool_ctx.info(f"Removing MCP server '{name}'")
|
|
|
|
# Check if server is running
|
|
if server.get("status") == "running" and server.get("process_id"):
|
|
if not force:
|
|
return f"Error: Server '{name}' is currently running. Use --force to remove anyway."
|
|
else:
|
|
# Stop the server process if it's running
|
|
process_id = server.get("process_id")
|
|
if process_id:
|
|
try:
|
|
import os
|
|
import signal
|
|
|
|
os.kill(process_id, signal.SIGTERM)
|
|
await tool_ctx.info(
|
|
f"Stopped running server '{name}' (PID: {process_id})"
|
|
)
|
|
except ProcessLookupError:
|
|
await tool_ctx.info(
|
|
f"Server '{name}' process not found (already stopped)"
|
|
)
|
|
|
|
# Remove from registry
|
|
del McpAddTool._mcp_servers[name]
|
|
McpAddTool._save_servers()
|
|
|
|
output = [
|
|
f"Successfully removed MCP server '{name}'",
|
|
f" Type: {server.get('type', 'unknown')}",
|
|
f" Command: {' '.join(server.get('command', []))}",
|
|
]
|
|
|
|
if server.get("tools"):
|
|
output.append(f" Tools removed: {len(server['tools'])}")
|
|
|
|
return "\n".join(output)
|
|
|
|
def register(self, mcp_server) -> None:
|
|
"""Register this tool with the MCP server."""
|
|
pass
|