haive.mcp.agents.mcp_agent¶

MCP Agent - Phase 4 Integration.

This agent demonstrates the complete MCP integration workflow: 1. Uses MCPManager to install and connect to MCP servers 2. Dynamically discovers and registers MCP tools 3. Integrates with Haive SimpleAgent for LLM-powered reasoning 4. Provides seamless tool execution through MCP protocol

Features: - Dynamic server discovery and installation - Automatic tool registration from MCP servers - Real LLM integration with structured output support - Multi-server coordination and tool management - Health monitoring and auto-reconnection

Usage:

from haive.mcp.agents.mcp_agent import MCPAgent from haive.core.engine.aug_llm import AugLLMConfig

# Create agent with automatic MCP integration agent = MCPAgent(

name=”research_assistant”, engine=AugLLMConfig(temperature=0.7), mcp_categories=[“development”, “productivity”] # Auto-install these categories

)

# Agent automatically installs MCP servers and registers tools result = await agent.arun(“Read the file ‘example.txt’ and search for Python tutorials”)

# Agent uses filesystem and search tools seamlessly

Classes¶

`MCPAgent`	Agent with seamless MCP integration.
`MCPIntegrationStats`	Statistics about MCP integration status.

Functions¶

`create_mcp_agent`([name, categories])	Factory function to create and initialize an MCP Agent.
`demo`()	Demo of MCP Agent.

Module Contents¶

class haive.mcp.agents.mcp_agent.MCPAgent¶

Bases: haive.agents.simple.agent.SimpleAgent

Agent with seamless MCP integration.

This agent extends SimpleAgent with automatic MCP server management, tool discovery, and seamless integration. It represents the culmination of Phase 4 - full agent integration with the MCP ecosystem.

The agent can: 1. Automatically install MCP servers from categories 2. Discover and register tools from connected servers 3. Use MCP tools transparently in LLM conversations 4. Monitor server health and auto-reconnect 5. Provide detailed integration statistics

Examples

Basic usage with automatic setup:

agent = MCPAgent(
    name="assistant",
    engine=AugLLMConfig(),
    mcp_categories=["development", "productivity"]
)

# Agent auto-installs filesystem, git, search tools
result = await agent.arun("List files and search for Python docs")

Custom server configuration:

agent = MCPAgent(
    name="custom_agent",
    engine=AugLLMConfig(),
    custom_servers={
        "database": MCPServerConfig(
            name="database",
            transport="stdio",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres"],
            env={"DATABASE_URL": "postgresql://..."}
        )
    }
)

With health monitoring:

agent = MCPAgent(
    name="monitored_agent",
    engine=AugLLMConfig(),
    auto_health_check=True,
    health_check_interval=30.0
)

# Get detailed integration statistics
stats = agent.get_mcp_stats()
print(f"Connected servers: {stats.servers_connected}")
print(f"Available tools: {stats.tools_registered}")

async arun(input_data, **kwargs)¶

Run the agent with automatic MCP initialization.

Parameters:: input_data (Any)
Return type:: Any

async discover_mcp_tools()¶

Discover tools from all connected MCP servers and register them.

Return type:: None

get_mcp_stats()¶

Get current MCP integration statistics.

Return type:: MCPIntegrationStats

async health_check_mcp()¶

Perform health check on all MCP servers.

Return type:: Dict[str, Any]

async initialize_mcp()¶

Initialize MCP integration - install servers and discover tools.

Return type:: None

async install_additional_category(category)¶

Install an additional MCP category after initialization.

Parameters:: category (str)
Return type:: bool

list_mcp_tools()¶

List all available MCP tools with details.

Return type:: List[Dict[str, Any]]

model_post_init(__context)¶

Initialize MCP components after Pydantic model creation.

Return type:: None

async refresh_mcp_tools()¶

Refresh tool discovery from all servers.

Return type:: None

run(input_data, **kwargs)¶

Sync run with MCP initialization.

Parameters:: input_data (Any)
Return type:: Any

class haive.mcp.agents.mcp_agent.MCPIntegrationStats(/, **data)¶

Bases: pydantic.BaseModel

Statistics about MCP integration status.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:: data (Any)

property connection_rate: float¶

Calculate server connection success rate.

Return type:: float

property tool_registration_rate: float¶

Calculate tool registration success rate.

Return type:: float

async haive.mcp.agents.mcp_agent.create_mcp_agent(name='mcp_agent', categories=None, **kwargs)¶

Factory function to create and initialize an MCP Agent.

Parameters:

name (str) – Agent name
categories (Optional[List[str]]) – MCP categories to install (defaults to [“development”, “productivity”])
**kwargs – Additional arguments for MCPAgent

Returns:

Fully initialized MCPAgent

Return type:

MCPAgent

async haive.mcp.agents.mcp_agent.demo()¶: Demo of MCP Agent.