mcp.agents.intelligent_mcp_agent ================================ .. py:module:: mcp.agents.intelligent_mcp_agent .. autoapi-nested-parse:: 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 .. rubric:: Example Dynamic server discovery and installation: .. code-block:: python 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 ---------- .. autoapisummary:: mcp.agents.intelligent_mcp_agent.logger Classes ------- .. autoapisummary:: mcp.agents.intelligent_mcp_agent.ApprovalStatus mcp.agents.intelligent_mcp_agent.HITLApprovalRequest mcp.agents.intelligent_mcp_agent.IntelligentMCPAgent mcp.agents.intelligent_mcp_agent.ServerRecommendation Functions --------- .. autoapisummary:: mcp.agents.intelligent_mcp_agent.create_agent_with_callback mcp.agents.intelligent_mcp_agent.create_auto_discovering_agent mcp.agents.intelligent_mcp_agent.create_manual_discovery_agent Module Contents --------------- .. py:class:: ApprovalStatus Bases: :py:obj:`str`, :py:obj:`enum.Enum` Status of HITL approval request. .. py:attribute:: APPROVED :value: 'approved' .. py:attribute:: PENDING :value: 'pending' .. py:attribute:: REJECTED :value: 'rejected' .. py:attribute:: TIMEOUT :value: 'timeout' .. py:class:: HITLApprovalRequest(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Request for human-in-the-loop approval. .. py:attribute:: approval_callback :type: str | None :value: None .. py:attribute:: context :type: dict[str, Any] :value: None .. py:attribute:: recommendation :type: ServerRecommendation :value: None .. py:attribute:: request_id :type: str :value: None .. py:attribute:: request_type :type: str :value: None .. py:attribute:: response_deadline :type: datetime.datetime | None :value: None .. py:attribute:: status :type: ApprovalStatus :value: None .. py:attribute:: timestamp :type: datetime.datetime :value: None .. py:class:: IntelligentMCPAgent(**kwargs) Bases: :py:obj:`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. .. attribute:: mcp_manager Dynamic MCP manager for server operations .. attribute:: doc_agent Documentation agent for server discovery .. attribute:: auto_discover Whether to automatically discover servers .. attribute:: require_approval Whether to require HITL approval .. attribute:: approval_timeout Timeout for approval requests in seconds .. attribute:: approval_callback Custom approval callback function .. attribute:: capability_cache Cache of discovered capabilities .. attribute:: recommendation_history History of recommendations .. rubric:: Example Basic intelligent agent setup: .. code-block:: python 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 .. py:method:: approve_request(request_id: str) -> bool :async: Approve a pending request. :param request_id: ID of the request to approve :returns: True if request was found and approved .. py:method:: arun(inputs: dict[str, Any]) -> Any :async: Run the agent with intelligent MCP discovery. Extends the base arun to add capability analysis and dynamic server installation based on user needs. :param inputs: Agent inputs with messages :returns: Agent response after processing with dynamic MCP .. py:method:: get_pending_approvals() -> list[HITLApprovalRequest] Get list of pending approval requests. .. py:method:: get_recommendation_history() -> list[ServerRecommendation] Get history of server recommendations. .. py:method:: reject_request(request_id: str) -> bool :async: Reject a pending request. :param request_id: ID of the request to reject :returns: True if request was found and rejected .. py:method:: setup() -> None :async: Setup the intelligent agent including documentation agent. .. py:attribute:: approval_callback :type: collections.abc.Callable | None :value: None .. py:attribute:: approval_timeout :type: float :value: None .. py:attribute:: auto_discover :type: bool :value: None .. py:attribute:: capability_cache :type: dict[str, list[str]] :value: None .. py:attribute:: doc_agent :type: Any | None :value: None .. py:attribute:: mcp_manager :type: haive.mcp.manager.MCPManager :value: None .. py:attribute:: pending_approvals :type: dict[str, HITLApprovalRequest] :value: None .. py:attribute:: recommendation_history :type: list[ServerRecommendation] :value: None .. py:attribute:: require_approval :type: bool :value: None .. py:class:: ServerRecommendation(/, **data: Any) Bases: :py:obj:`pydantic.BaseModel` Recommendation for MCP server installation. .. py:attribute:: alternative_servers :type: list[str] :value: None .. py:attribute:: capabilities :type: list[str] :value: None .. py:attribute:: confidence :type: float :value: None .. py:attribute:: config :type: haive.mcp.config.MCPServerConfig :value: None .. py:attribute:: reason :type: str :value: None .. py:attribute:: server_name :type: str :value: None .. py:function:: create_agent_with_callback(engine: haive.core.engine.aug_llm.AugLLMConfig, approval_callback: collections.abc.Callable[[HITLApprovalRequest], bool]) -> IntelligentMCPAgent :async: Create an agent with custom approval callback. :param engine: LLM engine configuration :param approval_callback: Custom function to handle approvals :returns: Configured IntelligentMCPAgent .. py:function:: 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. :param engine: LLM engine configuration :param require_approval: Whether to require HITL approval :returns: Configured IntelligentMCPAgent .. py:function:: create_manual_discovery_agent(engine: haive.core.engine.aug_llm.AugLLMConfig) -> IntelligentMCPAgent Create an agent with manual discovery tools but no auto-discovery. :param engine: LLM engine configuration :returns: Configured IntelligentMCPAgent .. py:data:: logger