API Reference

mcp

MCP integration for Haystack

Module haystack_integrations.tools.mcp.mcp_tool

MCPError

Base class for MCP-related errors.

MCPError.__init__

def __init__(message: str) -> None

Initialize the MCPError.

Arguments:

  • message: Descriptive error message

MCPConnectionError

Error connecting to MCP server.

MCPConnectionError.__init__

def __init__(message: str,
             server_info: "MCPServerInfo | None" = None,
             operation: str | None = None) -> None

Initialize the MCPConnectionError.

Arguments:

  • message: Descriptive error message
  • server_info: Server connection information that was used
  • operation: Name of the operation that was being attempted

MCPToolNotFoundError

Error when a tool is not found on the server.

MCPToolNotFoundError.__init__

def __init__(message: str,
             tool_name: str,
             available_tools: list[str] | None = None) -> None

Initialize the MCPToolNotFoundError.

Arguments:

  • message: Descriptive error message
  • tool_name: Name of the tool that was requested but not found
  • available_tools: List of available tool names, if known

MCPResponseTypeError

Error when response content type is not supported.

MCPResponseTypeError.__init__

def __init__(message: str,
             response: Any,
             tool_name: str | None = None) -> None

Initialize the MCPResponseTypeError.

Arguments:

  • message: Descriptive error message
  • response: The response that had the wrong type
  • tool_name: Name of the tool that produced the response

MCPInvocationError

Error during tool invocation.

MCPInvocationError.__init__

def __init__(message: str,
             tool_name: str,
             tool_args: dict[str, Any] | None = None) -> None

Initialize the MCPInvocationError.

Arguments:

  • message: Descriptive error message
  • tool_name: Name of the tool that was being invoked
  • tool_args: Arguments that were passed to the tool

MCPClient

Abstract base class for MCP clients.

This class defines the common interface and shared functionality for all MCP clients, regardless of the transport mechanism used.

MCPClient.connect

@abstractmethod
async def connect() -> list[Tool]

Connect to an MCP server.

Raises:

  • MCPConnectionError: If connection to the server fails

Returns:

List of available tools on the server

MCPClient.call_tool

async def call_tool(tool_name: str, tool_args: dict[str, Any]) -> Any

Call a tool on the connected MCP server.

Arguments:

  • tool_name: Name of the tool to call
  • tool_args: Arguments to pass to the tool

Raises:

  • MCPConnectionError: If not connected to an MCP server
  • MCPInvocationError: If the tool invocation fails
  • MCPResponseTypeError: If response type is not TextContent

Returns:

Result of the tool invocation

MCPClient.close

async def close() -> None

Close the connection and clean up resources.

This method ensures all resources are properly released, even if errors occur.

StdioClient

MCP client that connects to servers using stdio transport.

StdioClient.__init__

def __init__(command: str,
             args: list[str] | None = None,
             env: dict[str, str] | None = None) -> None

Initialize a stdio MCP client.

Arguments:

  • command: Command to run (e.g., "python", "node")
  • args: Arguments to pass to the command
  • env: Environment variables for the command

StdioClient.connect

async def connect() -> list[Tool]

Connect to an MCP server using stdio transport.

Raises:

  • MCPConnectionError: If connection to the server fails

Returns:

List of available tools on the server

SSEClient

MCP client that connects to servers using SSE transport.

SSEClient.__init__

def __init__(base_url: str,
             token: str | None = None,
             timeout: int = 5) -> None

Initialize an SSE MCP client.

Arguments:

  • base_url: Base URL of the server
  • token: Authentication token for the server (optional)
  • timeout: Connection timeout in seconds

SSEClient.connect

async def connect() -> list[Tool]

Connect to an MCP server using SSE transport.

Raises:

  • MCPConnectionError: If connection to the server fails

Returns:

List of available tools on the server

MCPServerInfo

Abstract base class for MCP server connection parameters.

This class defines the common interface for all MCP server connection types.

MCPServerInfo.create_client

@abstractmethod
def create_client() -> MCPClient

Create an appropriate MCP client for this server info.

Returns:

An instance of MCPClient configured with this server info

MCPServerInfo.to_dict

def to_dict() -> dict[str, Any]

Serialize this server info to a dictionary.

Returns:

Dictionary representation of this server info

MCPServerInfo.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "MCPServerInfo"

Deserialize server info from a dictionary.

Arguments:

  • data: Dictionary containing serialized server info

Returns:

Instance of the appropriate server info class

SSEServerInfo

Data class that encapsulates SSE MCP server connection parameters.

Arguments:

  • base_url: Base URL of the MCP server
  • token: Authentication token for the server (optional)
  • timeout: Connection timeout in seconds

SSEServerInfo.create_client

def create_client() -> MCPClient

Create an SSE MCP client.

Returns:

Configured HttpMCPClient instance

StdioServerInfo

Data class that encapsulates stdio MCP server connection parameters.

Arguments:

  • command: Command to run (e.g., "python", "node")
  • args: Arguments to pass to the command
  • env: Environment variables for the command

StdioServerInfo.create_client

def create_client() -> MCPClient

Create a stdio MCP client.

Returns:

Configured StdioMCPClient instance

MCPTool

A Tool that represents a single tool from an MCP server.

This implementation uses the official MCP SDK for protocol handling while maintaining compatibility with the Haystack tool ecosystem.

Response handling:

  • Text content is supported and returned as strings
  • Unsupported content types (like binary/images) will raise MCPResponseTypeError

Example using HTTP:

from haystack.tools import MCPTool, SSEServerInfo

# Create tool instance
tool = MCPTool(
    name="add",
    server_info=SSEServerInfo(base_url="http://localhost:8000")
)

# Use the tool
result = tool.invoke(a=5, b=3)

Example using stdio:

from haystack.tools import MCPTool, StdioServerInfo

# Create tool instance
tool = MCPTool(
    name="get_current_time",
    server_info=StdioServerInfo(command="python", args=["path/to/server.py"])
)

# Use the tool
result = tool.invoke(timezone="America/New_York")

MCPTool.__init__

def __init__(name: str,
             server_info: MCPServerInfo,
             description: str | None = None,
             connection_timeout: int = 30,
             invocation_timeout: int = 30)

Initialize the MCP tool.

Arguments:

  • name: Name of the tool to use
  • server_info: Server connection information
  • description: Custom description (if None, server description will be used)
  • connection_timeout: Timeout in seconds for server connection
  • invocation_timeout: Default timeout in seconds for tool invocations

Raises:

  • MCPConnectionError: If connection to the server fails
  • MCPToolNotFoundError: If no tools are available or the requested tool is not found
  • TimeoutError: If connection times out

MCPTool.ainvoke

async def ainvoke(**kwargs: Any) -> Any

Asynchronous tool invocation.

Arguments:

  • kwargs: Arguments to pass to the tool

Raises:

  • MCPInvocationError: If the tool invocation fails
  • MCPResponseTypeError: If response type is not supported
  • TimeoutError: If the operation times out

Returns:

Result of the tool invocation, processed based on content type

MCPTool.to_dict

def to_dict() -> dict[str, Any]

Serializes the MCPTool to a dictionary.

The serialization preserves all information needed to recreate the tool, including server connection parameters and timeout settings. Note that the active connection is not maintained.

Returns:

Dictionary with serialized data in the format: {"type": fully_qualified_class_name, "data": {parameters}}

MCPTool.from_dict

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Tool"

Deserializes the MCPTool from a dictionary.

This method reconstructs an MCPTool instance from a serialized dictionary, including recreating the server_info object. A new connection will be established to the MCP server during initialization.

Arguments:

  • data: Dictionary containing serialized tool data

Raises:

  • None: Various exceptions if connection fails

Returns:

A fully initialized MCPTool instance