aiq.tool.mcp.mcp_client

Attributes

logger

Classes

MCPSSEClient	Client for creating a session and connecting to an MCP server using SSE
MCPBuilder	Builder class used to connect to an MCP Server and generate ToolClients
MCPToolClient	Client wrapper used to call an MCP tool.

Functions

model_from_mcp_schema(→ type[pydantic.BaseModel])

Create a pydantic model from the input schema of the MCP tool

Module Contents

logger

model_from_mcp_schema(

name: str,

mcp_input_schema: dict,

) → type[pydantic.BaseModel]

Create a pydantic model from the input schema of the MCP tool

class MCPSSEClient(url: str)

Client for creating a session and connecting to an MCP server using SSE

Args:

url (str): The url of the MCP server

url

async connect_to_sse_server()

Establish a session with an MCP SSE server within an aync context

class MCPBuilder(url)

Bases: MCPSSEClient

Builder class used to connect to an MCP Server and generate ToolClients

Args:

url (str): The url of the MCP server

_tools = None

async get_tools()

Retrieve a dictionary of all tools served by the MCP server.

async get_tool(tool_name: str) → MCPToolClient

Get an MCP Tool by name.

Args:

tool_name (str): Name of the tool to load.

Returns:

MCPToolClient for the configured tool.

Raise:

ValueError if no tool is available with that name.

async call_tool(tool_name: str, tool_args: dict | None)

class MCPToolClient(

url: str,

tool_name: str,

tool_description: str | None,

tool_input_schema: dict | None = None,

)

Bases: MCPSSEClient

Client wrapper used to call an MCP tool.

Args:

url (str): The url of the MCP server tool_name (str): The name of the tool to wrap tool_description (str): The description of the tool provided by the MCP server. tool_input_schema (dict): The input schema for the tool.

_tool_name

_tool_description

_input_schema = None

property name

Returns the name of the tool.

property description

Returns the tool’s description. If none was provided. Provides a simple description using the tool’s name

property input_schema

Returns the tool’s input_schema.

set_description(description: str)

Manually define the tool’s description using the provided string.

async acall(tool_args: dict) → str

Call the MCP tool with the provided arguments.

Args:

tool_args (dict[str, Any]): A dictionary of key value pairs to serve as inputs for the MCP tool.