Function execute_tool

optillm/plugins/mcp_plugin.py:664–703

Execute a tool on an MCP server This function creates a fresh connection for each tool execution to ensure reliability.

(server_name: str, tool_name: str, arguments: Dict[str, Any])

        return {"error": f"Error executing tool: {str(e)}"}

async def execute_tool(server_name: str, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """
    Execute a tool on an MCP server

    This function creates a fresh connection for each tool execution to ensure reliability.
    """
    logger.info(f"Executing tool {tool_name} on server {server_name} with arguments: {arguments}")

    # Load configuration
    config_manager = MCPConfigManager()
    if not config_manager.load_config():
        return {"error": "Failed to load MCP configuration"}

    # Get server configuration
    server_config = config_manager.servers.get(server_name)
    if not server_config:
        return {"error": f"Server {server_name} not found in configuration"}

    # Log the tool call in detail
    logger.debug(f"Tool call details:")
    logger.debug(f"  Server: {server_name}")
    logger.debug(f"  Tool: {tool_name}")
    logger.debug(f"  Arguments: {json.dumps(arguments, indent=2)}")
    logger.debug(f"  Transport: {server_config.transport}")

    try:
        # Route to appropriate transport
        if server_config.transport == "stdio":
            return await execute_tool_stdio(server_config, tool_name, arguments)
        elif server_config.transport == "sse":
            return await execute_tool_sse(server_config, tool_name, arguments)
        elif server_config.transport == "websocket":
            return await execute_tool_websocket(server_config, tool_name, arguments)
        else:
            return {"error": f"Unsupported transport type: {server_config.transport}"}

    except Exception as e:
        logger.error(f"Error executing tool {tool_name} on server {server_name}: {e}")
        logger.error(traceback.format_exc())
        return {"error": f"Error executing tool: {str(e)}"}

async def execute_tool_stdio(server_config: ServerConfig, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Execute tool using stdio transport"""