surgeon

SurgeonBackend

Source code in wintermute/integrations/surgeon/backend.py

class SurgeonBackend:
    # [FIX] Add return type annotation
    def __init__(self, surgeon_root: str) -> None:
        current_dir = Path(__file__).parent
        server_script = current_dir / "server.py"

        self.controller = SurgeonController(str(server_script), surgeon_root)
        self.session: Optional[ClientSession] = None
        self._exit_stack: Optional[AsyncExitStack] = None

    async def start(self) -> None:
        """Starts the subprocess and initializes the MCP session."""
        if not self.controller.start():
            raise RuntimeError("Failed to start SURGEON MCP server.")

        server_params = StdioServerParameters(
            command="python3",
            args=[str(self.controller.mcp_script)],
            env={"SURGEON_ROOT": str(self.controller.surgeon_root)},
        )

        self._exit_stack = AsyncExitStack()

        try:
            read, write = await self._exit_stack.enter_async_context(
                stdio_client(server_params)
            )
            self.session = await self._exit_stack.enter_async_context(
                ClientSession(read, write)
            )
            await self.session.initialize()
            log.info("Connected to SURGEON MCP Session.")

        except Exception as e:
            log.error(f"Failed to connect MCP session: {e}")
            await self.stop()

    # [FIX] Add return type annotation
    async def stop(self) -> None:
        if self._exit_stack:
            await self._exit_stack.aclose()
        self.controller.stop()

    async def get_ai_tools(self) -> List[Dict[str, Any]]:
        """
        Fetches tools from MCP and converts them to OpenAI/Wintermute format.
        """
        if not self.session:
            return []

        try:
            mcp_tools = await self.session.list_tools()
            openai_tools = []

            for tool in mcp_tools.tools:
                openai_tools.append(
                    {
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": tool.inputSchema,
                        },
                    }
                )

            return openai_tools
        except Exception as e:
            log.error(f"Failed to list tools from SURGEON MCP: {e}")
            return []

    async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        """
        Executes a tool on the remote MCP server.
        """
        if not self.session:
            return "Error: SURGEON Backend not connected."

        try:
            result = await self.session.call_tool(tool_name, arguments)
            return "\n".join([c.text for c in result.content if c.type == "text"])
        except Exception as e:
            log.error(f"SURGEON Execution Error: {e}")
            return f"Error executing {tool_name}: {str(e)}"

execute_tool(tool_name, arguments) async

Executes a tool on the remote MCP server.

Source code in wintermute/integrations/surgeon/backend.py

async def execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
    """
    Executes a tool on the remote MCP server.
    """
    if not self.session:
        return "Error: SURGEON Backend not connected."

    try:
        result = await self.session.call_tool(tool_name, arguments)
        return "\n".join([c.text for c in result.content if c.type == "text"])
    except Exception as e:
        log.error(f"SURGEON Execution Error: {e}")
        return f"Error executing {tool_name}: {str(e)}"

get_ai_tools() async

Fetches tools from MCP and converts them to OpenAI/Wintermute format.

Source code in wintermute/integrations/surgeon/backend.py

async def get_ai_tools(self) -> List[Dict[str, Any]]:
    """
    Fetches tools from MCP and converts them to OpenAI/Wintermute format.
    """
    if not self.session:
        return []

    try:
        mcp_tools = await self.session.list_tools()
        openai_tools = []

        for tool in mcp_tools.tools:
            openai_tools.append(
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema,
                    },
                }
            )

        return openai_tools
    except Exception as e:
        log.error(f"Failed to list tools from SURGEON MCP: {e}")
        return []

start() async

Starts the subprocess and initializes the MCP session.

Source code in wintermute/integrations/surgeon/backend.py

async def start(self) -> None:
    """Starts the subprocess and initializes the MCP session."""
    if not self.controller.start():
        raise RuntimeError("Failed to start SURGEON MCP server.")

    server_params = StdioServerParameters(
        command="python3",
        args=[str(self.controller.mcp_script)],
        env={"SURGEON_ROOT": str(self.controller.surgeon_root)},
    )

    self._exit_stack = AsyncExitStack()

    try:
        read, write = await self._exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.session = await self._exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        await self.session.initialize()
        log.info("Connected to SURGEON MCP Session.")

    except Exception as e:
        log.error(f"Failed to connect MCP session: {e}")
        await self.stop()