@classmethod
def get_instance(cls) -> "AsyncExecutor"

def __init__()

def run(coro: Coroutine[Any, Any, Any], timeout: float | None = None) -> Any

def get_loop()

def run_background(
    coro_factory: Callable[[asyncio.Event], Coroutine[Any, Any, Any]],
    timeout: float | None = None
) -> tuple[concurrent.futures.Future[Any], asyncio.Event]

def shutdown(timeout: float = 2) -> None

def __init__(message: str) -> None

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

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

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

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

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

async def aclose() -> None

def __init__(command: str,
             args: list[str] | None = None,
             env: dict[str, str | Secret] | None = None,
             max_retries: int = 3,
             base_delay: float = 1.0,
             max_delay: float = 30.0) -> None

async def connect() -> list[types.Tool]

def __init__(server_info: "SSEServerInfo",
             max_retries: int = 3,
             base_delay: float = 1.0,
             max_delay: float = 30.0) -> None

async def connect() -> list[types.Tool]

def __init__(server_info: "StreamableHttpServerInfo",
             max_retries: int = 3,
             base_delay: float = 1.0,
             max_delay: float = 30.0) -> None

async def connect() -> list[types.Tool]

@abstractmethod
def create_client() -> MCPClient

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

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

server_info = SSEServerInfo(
    url="https://my-mcp-server.com",
    token=Secret.from_env_var("API_KEY"),
)

def __post_init__()

def create_client() -> MCPClient

server_info = StreamableHttpServerInfo(
    url="https://my-mcp-server.com",
    token=Secret.from_env_var("API_KEY"),
)

def __post_init__()

def create_client() -> MCPClient

server_info = StdioServerInfo(
    command="uv",
    args=["run", "my-mcp-server"],
    env={
        "WORKSPACE_PATH": "/path/to/workspace",  # Plain string
        "API_KEY": Secret.from_env_var("API_KEY"),  # Secret object
    }
)

def create_client() -> MCPClient

import json
from haystack_integrations.tools.mcp import MCPTool, StreamableHttpServerInfo

# Create tool instance
tool = MCPTool(
    name="multiply",
    server_info=StreamableHttpServerInfo(url="https://:8000/mcp")
)

# Use the tool and parse result
result_json = tool.invoke(a=5, b=3)
result = json.loads(result_json)

import json
from haystack.tools import MCPTool, SSEServerInfo

# Create tool instance
tool = MCPTool(
    name="add",
    server_info=SSEServerInfo(url="https://:8000/sse")
)

# Use the tool and parse result
result_json = tool.invoke(a=5, b=3)
result = json.loads(result_json)

import json
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 and parse result
result_json = tool.invoke(timezone="America/New_York")
result = json.loads(result_json)

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

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

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

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

def close()

def __del__()

def tools() -> list[types.Tool]

def stop() -> None

# Prerequisites:
# 1. pip install uvx mcp-server-time  # Install required MCP server and tools
# 2. export OPENAI_API_KEY="your-api-key"  # Set up your OpenAI API key

import os
from haystack import Pipeline
from haystack.components.converters import OutputAdapter
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.tools import ToolInvoker
from haystack.dataclasses import ChatMessage
from haystack_integrations.tools.mcp import MCPToolset, StdioServerInfo

# Create server info for the time service (can also use SSEServerInfo for remote servers)
server_info = StdioServerInfo(command="uvx", args=["mcp-server-time", "--local-timezone=Europe/Berlin"])

# Create the toolset - this will automatically discover all available tools
# You can optionally specify which tools to include
mcp_toolset = MCPToolset(
    server_info=server_info,
    tool_names=["get_current_time"]  # Only include the get_current_time tool
)

# Create a pipeline with the toolset
pipeline = Pipeline()
pipeline.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini", tools=mcp_toolset))
pipeline.add_component("tool_invoker", ToolInvoker(tools=mcp_toolset))
pipeline.add_component(
    "adapter",
    OutputAdapter(
        template="{{ initial_msg + initial_tool_messages + tool_messages }}",
        output_type=list[ChatMessage],
        unsafe=True,
    ),
)
pipeline.add_component("response_llm", OpenAIChatGenerator(model="gpt-4o-mini"))
pipeline.connect("llm.replies", "tool_invoker.messages")
pipeline.connect("llm.replies", "adapter.initial_tool_messages")
pipeline.connect("tool_invoker.tool_messages", "adapter.tool_messages")
pipeline.connect("adapter.output", "response_llm.messages")

# Run the pipeline with a user question
user_input = "What is the time in New York? Be brief."
user_input_msg = ChatMessage.from_user(text=user_input)

result = pipeline.run({"llm": {"messages": [user_input_msg]}, "adapter": {"initial_msg": [user_input_msg]}})
print(result["response_llm"]["replies"][0].text)

from haystack_integrations.tools.mcp import MCPToolset, StreamableHttpServerInfo

# Create the toolset with streamable HTTP connection
toolset = MCPToolset(
    server_info=StreamableHttpServerInfo(url="https://:8000/mcp"),
    tool_names=["multiply"]  # Optional: only include specific tools
)
# Use the toolset as shown in the pipeline example above

from haystack_integrations.tools.mcp import MCPToolset, SSEServerInfo
from haystack.components.tools import ToolInvoker

# Create the toolset with an SSE connection
sse_toolset = MCPToolset(
    server_info=SSEServerInfo(url="http://some-remote-server.com:8000/sse"),
    tool_names=["add", "subtract"]  # Only include specific tools
)

# Use the toolset as shown in the pipeline example above

def __init__(server_info: MCPServerInfo,
             tool_names: list[str] | None = None,
             connection_timeout: float = 30.0,
             invocation_timeout: float = 30.0)

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

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

def close()