mcp.agents.intelligent_mcp_agentΒΆ
Intelligent MCP Agent for dynamic server discovery and management.
This module provides an advanced agent that can dynamically discover, recommend, install, and manage MCP servers based on user needs. It includes HITL (Human-In-The-Loop) approval workflows and intelligent capability matching.
- The agent supports:
Dynamic server discovery from 992+ server database
Intelligent capability matching
HITL approval workflows
Hot-reload and dynamic tool refresh
Server health monitoring
Automatic retry and recovery
- Classes:
IntelligentMCPAgent: Advanced agent for dynamic MCP management HITLApprovalRequest: Request structure for human approval ServerRecommendation: Recommendation for MCP server installation
Example
Dynamic server discovery and installation:
from haive.mcp.agents import IntelligentMCPAgent
from haive import core from haive import agents
# Create intelligent agent agent = IntelligentMCPAgent(
engine=engine, auto_discover=True, require_approval=True
)
# Agent discovers need for database capabilities result = await agent.arun({
- βmessagesβ: [{
βroleβ: βuserβ, βcontentβ: βI need to analyze data from PostgreSQLβ
}]
})
# Agent recommends postgres MCP server and asks for approval # After approval, installs and configures automatically
AttributesΒΆ
ClassesΒΆ
Status of HITL approval request. |
|
Request for human-in-the-loop approval. |
|
Intelligent agent for dynamic MCP server discovery and management. |
|
Recommendation for MCP server installation. |
FunctionsΒΆ
|
Create an agent with custom approval callback. |
|
Create an agent that automatically discovers and installs MCP servers. |
|
Create an agent with manual discovery tools but no auto-discovery. |
Module ContentsΒΆ
- class mcp.agents.intelligent_mcp_agent.ApprovalStatusΒΆ
-
Status of HITL approval request.
- APPROVED = 'approved'ΒΆ
- PENDING = 'pending'ΒΆ
- REJECTED = 'rejected'ΒΆ
- TIMEOUT = 'timeout'ΒΆ
- class mcp.agents.intelligent_mcp_agent.HITLApprovalRequest(/, **data: Any)ΒΆ
Bases:
pydantic.BaseModel
Request for human-in-the-loop approval.
- recommendation: ServerRecommendation = NoneΒΆ
- response_deadline: datetime.datetime | None = NoneΒΆ
- status: ApprovalStatus = NoneΒΆ
- timestamp: datetime.datetime = NoneΒΆ
- class mcp.agents.intelligent_mcp_agent.IntelligentMCPAgent(**kwargs)ΒΆ
Bases:
haive.agents.react.ReactAgent
Intelligent agent for dynamic MCP server discovery and management.
This agent extends ReactAgent with intelligent MCP management capabilities, including dynamic server discovery, HITL approval workflows, and automatic configuration based on user needs.
- mcp_managerΒΆ
Dynamic MCP manager for server operations
- doc_agentΒΆ
Documentation agent for server discovery
- auto_discoverΒΆ
Whether to automatically discover servers
- require_approvalΒΆ
Whether to require HITL approval
- approval_timeoutΒΆ
Timeout for approval requests in seconds
- approval_callbackΒΆ
Custom approval callback function
- capability_cacheΒΆ
Cache of discovered capabilities
- recommendation_historyΒΆ
History of recommendations
Example
Basic intelligent agent setup:
agent = IntelligentMCPAgent( engine=engine, auto_discover=True, require_approval=True, approval_timeout=30.0 ) # Agent automatically discovers and recommends servers result = await agent.arun({ "messages": [{ "role": "user", "content": "Help me search the web for Python tutorials" }] }) # Agent detects need for web search, recommends brave-search server # Waits for approval, then installs and uses it
- async approve_request(request_id: str) bool ΒΆ
Approve a pending request.
- Parameters:
request_id β ID of the request to approve
- Returns:
True if request was found and approved
- async arun(inputs: dict[str, Any]) Any ΒΆ
Run the agent with intelligent MCP discovery.
Extends the base arun to add capability analysis and dynamic server installation based on user needs.
- Parameters:
inputs β Agent inputs with messages
- Returns:
Agent response after processing with dynamic MCP
- get_pending_approvals() list[HITLApprovalRequest] ΒΆ
Get list of pending approval requests.
- get_recommendation_history() list[ServerRecommendation] ΒΆ
Get history of server recommendations.
- async reject_request(request_id: str) bool ΒΆ
Reject a pending request.
- Parameters:
request_id β ID of the request to reject
- Returns:
True if request was found and rejected
- approval_callback: collections.abc.Callable | None = NoneΒΆ
- mcp_manager: haive.mcp.manager.MCPManager = NoneΒΆ
- pending_approvals: dict[str, HITLApprovalRequest] = NoneΒΆ
- recommendation_history: list[ServerRecommendation] = NoneΒΆ
- class mcp.agents.intelligent_mcp_agent.ServerRecommendation(/, **data: Any)ΒΆ
Bases:
pydantic.BaseModel
Recommendation for MCP server installation.
- config: haive.mcp.config.MCPServerConfig = NoneΒΆ
- async mcp.agents.intelligent_mcp_agent.create_agent_with_callback(engine: haive.core.engine.aug_llm.AugLLMConfig, approval_callback: collections.abc.Callable[[HITLApprovalRequest], bool]) IntelligentMCPAgent ΒΆ
Create an agent with custom approval callback.
- Parameters:
engine β LLM engine configuration
approval_callback β Custom function to handle approvals
- Returns:
Configured IntelligentMCPAgent
- mcp.agents.intelligent_mcp_agent.create_auto_discovering_agent(engine: haive.core.engine.aug_llm.AugLLMConfig, require_approval: bool = True) IntelligentMCPAgent ΒΆ
Create an agent that automatically discovers and installs MCP servers.
- Parameters:
engine β LLM engine configuration
require_approval β Whether to require HITL approval
- Returns:
Configured IntelligentMCPAgent
- mcp.agents.intelligent_mcp_agent.create_manual_discovery_agent(engine: haive.core.engine.aug_llm.AugLLMConfig) IntelligentMCPAgent ΒΆ
Create an agent with manual discovery tools but no auto-discovery.
- Parameters:
engine β LLM engine configuration
- Returns:
Configured IntelligentMCPAgent
- mcp.agents.intelligent_mcp_agent.loggerΒΆ