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

Attributes¶

logger

Classes¶

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

Functions¶

`create_mcp_agent`(→ MCPAgent)	Factory function to create and initialize an MCP Agent.
`demo`()	Demo of MCP Agent.

Module Contents¶

class mcp.agents.mcp_agent.MCPAgent(/, **data: Any)¶

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: Any, **kwargs) → Any¶: Run the agent with automatic MCP initialization.

async discover_mcp_tools() → None¶: Discover tools from all connected MCP servers and register them.

get_mcp_stats() → MCPIntegrationStats¶: Get current MCP integration statistics.

async health_check_mcp() → Dict[str, Any]¶: Perform health check on all MCP servers.

async initialize_mcp() → None¶: Initialize MCP integration - install servers and discover tools.

async install_additional_category(category: str) → bool¶: Install an additional MCP category after initialization.

list_mcp_tools() → List[Dict[str, Any]]¶: List all available MCP tools with details.

model_post_init(__context) → None¶: Initialize MCP components after Pydantic model creation.

async refresh_mcp_tools() → None¶: Refresh tool discovery from all servers.

run(input_data: Any, **kwargs) → Any¶: Sync run with MCP initialization.

auto_health_check: bool = None¶

auto_install: bool = None¶

custom_servers: Dict[str, haive.mcp.manager.MCPServerConfig] | None = None¶

health_check_interval: float = None¶

max_concurrent_installs: int = None¶

mcp_categories: List[str] | None = None¶

mcp_manager: haive.mcp.manager.MCPManager | None = None¶

mcp_stats: MCPIntegrationStats | None = None¶

mcp_tools: Dict[str, langchain_core.tools.Tool] | None = None¶

server_tools_map: Dict[str, List[str]] | None = None¶

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

Bases: pydantic.BaseModel

Statistics about MCP integration status.

categories_active: List[str] = None¶

property connection_rate: float¶: Calculate server connection success rate.

last_discovery: datetime.datetime | None = None¶

servers_connected: int = None¶

servers_installed: int = None¶

property tool_registration_rate: float¶: Calculate tool registration success rate.

tools_discovered: int = None¶

tools_registered: int = None¶

async mcp.agents.mcp_agent.create_mcp_agent(name: str = 'mcp_agent', categories: List[str] | None = None, **kwargs) → MCPAgent¶

Factory function to create and initialize an MCP Agent.

Parameters:

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

Returns:

Fully initialized MCPAgent

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

mcp.agents.mcp_agent.logger¶