mcp.mixins.mcp_mixin

MCP mixin for adding Model Context Protocol capabilities to agents.

This module provides a mixin class that adds MCP functionality to any Haive agent. The mixin handles server connections, tool discovery, resource access, and prompt management with automatic error handling and graceful degradation.

The MCPMixin class integrates seamlessly with the Haive agent architecture, providing:

  • Dynamic server discovery and connection management

  • Automatic tool registration with agents

  • Health monitoring and failure recovery

  • Component registry integration

  • Lazy initialization support

Examples

Adding MCP capabilities to a custom agent:

from haive.agents.base import Agent
from haive.mcp.mixins import MCPMixin
from haive.mcp.config import MCPConfig

class MyCustomAgent(MCPMixin, Agent):
    '''Agent with MCP capabilities.'''

    async def setup(self):
        await super().setup()
        # MCP tools are now available
        if self._mcp_tools:
            print(f"Loaded {len(self._mcp_tools)} MCP tools")

# Use the agent
agent = MyCustomAgent(
    engine=engine,
    mcp_config=MCPConfig(enabled=True, servers={...})
)

See also

Attributes

MCP_AVAILABLE

logger

Classes

MCPMixin

Mixin to add MCP (Model Context Protocol) capabilities to any agent.

Module Contents

class mcp.mixins.mcp_mixin.MCPMixin(/, **data: Any)

Bases: pydantic.BaseModel

Mixin to add MCP (Model Context Protocol) capabilities to any agent.

This mixin provides comprehensive MCP integration including server management, tool discovery, health monitoring, and graceful error handling. It can be mixed into any Haive agent to add MCP functionality.

mcp_config

Optional MCP configuration for server connections

Private Attributes:

_mcp_client: The MultiServerMCPClient instance for server communication _mcp_servers: Dictionary of configured MCP servers _mcp_tools: Dictionary of discovered tools from MCP servers _mcp_initialized: Flag indicating if MCP has been initialized _failed_servers: Set of server names that failed to connect _server_health: Health status tracking for each server

Features:

  • Dynamic MCP server discovery and connection

  • Automatic tool registration with agent systems

  • Health monitoring with automatic reconnection

  • Graceful degradation when servers fail

  • Lazy initialization support

  • Resource and prompt management

Examples

Creating a custom agent with MCP capabilities:

from haive.agents.base import Agent
from haive.mcp.mixins import MCPMixin
from haive.mcp.config import MCPConfig, MCPServerConfig

class MyMCPAgent(MCPMixin, Agent):
    '''Custom agent with MCP capabilities.'''

    async def setup(self):
        await super().setup()
        # Initialize MCP
        if self.mcp_config:
            await self.initialize_mcp()
            print(f"Loaded {len(self._mcp_tools)} MCP tools")

# Configure and use
agent = MyMCPAgent(
    engine=engine,
    mcp_config=MCPConfig(
        enabled=True,
        servers={
            "filesystem": MCPServerConfig(
                name="filesystem",
                transport="stdio",
                command="npx",
                args=["-y", "@modelcontextprotocol/server-filesystem"]
            )
        }
    )
async call_mcp_tool(tool_name: str, arguments: dict[str, Any], server_name: str | None = None) → Any

Call an MCP tool.

Parameters:

  • tool_name – Name of the tool to call

  • arguments – Tool arguments

  • server_name – Optional specific server to use

Returns:

Tool execution result

async cleanup_mcp()

Clean up MCP resources.

async get_mcp_prompt(server_name: str, prompt_name: str, arguments: dict[str, Any] | None = None) → list[Any]

Get a prompt from an MCP server.

async get_mcp_resources(server_name: str, uris: str | list[str] | None = None) → list[Any]

Get resources from an MCP server.

get_mcp_status() → dict[str, Any]

Get current MCP status.

async initialize_mcp() → bool

Initialize MCP servers and client.

Establishes connections to all configured MCP servers, discovers available tools, and registers them with the component registry if available.

The initialization process:

  1. Checks for MCP dependencies availability

  2. Discovers servers if auto-discovery is enabled

  3. Connects to each configured server

  4. Creates MultiServerMCPClient with connected servers

  5. Discovers tools from all servers

  6. Registers servers and tools with component registry

Returns:

True if at least one server connected successfully, False otherwise

Return type:

bool

Raises:

Exception – Logged but not raised to ensure graceful degradation

Examples

Manual initialization:

agent = MCPAgent(engine=engine, mcp_config=config)
success = await agent.initialize_mcp()
if success:
    print(f"Connected to {len(agent._mcp_servers)} servers")
async mcp_session(server_name: str | None = None)

Context manager for MCP operations.

Parameters:

server_name – Optional specific server to use

Yields:

MCP client or session

model_post_init(__context) → None

Initialize MCP mixin after model initialization.

async refresh_mcp_servers()

Refresh MCP server connections.

setup_mcp()

Setup MCP after agent initialization.

mcp_config: haive.mcp.config.MCPConfig | None = None
mcp.mixins.mcp_mixin.MCP_AVAILABLE = True
mcp.mixins.mcp_mixin.logger