AgenticMaid Project

AgenticMaid 项目

概述

AgenticMaid 是一个 Python 库，旨在与一个或多个多能力协议（MCP）服务交互。它支持动态获取和使用这些服务提供的工具（能力）。客户端还可以管理各种 AI/大模型服务的配置、安排自动化任务并处理聊天服务交互，因此是构建 AI 驱动应用的多功能组件。

它使用 langchain-mcp-adapters 与 MCP 服务通信，并使用 langgraph 创建能够使用已获取 MCP 工具的响应式 Agent。

功能

• 多服务 MCP 交互： 连接多个 MCP 服务并使用其工具。
• 动态获取工具： 在运行时从 MCP 服务获取可用工具。
• 灵活配置： 支持通过 Python 字典、JSON 文件和 .env 文件配置敏感数据。
• AI 服务管理： 配置并使用各种 AI/大模型服务（例如 OpenAI、Anthropic、Azure OpenAI、本地模型）。
• 计划任务： 支持定义并运行基于类似 cron 的计划安排的任务。
• 聊天服务集成： 提供用于处理与已定义聊天服务交互的框架。
• 创建 Agent： 使用 langgraph 创建可利用 MCP 工具和已配置大模型的 ReAct Agent。
• 环境变量支持： 从 .env 文件加载默认配置和敏感密钥（如 API 密钥）。

安装

1. 前提条件：

   • Python 3.8+

2. 克隆仓库（如适用），或将 AgenticMaid 添加到项目中。

3. 安装依赖项： 客户端依赖多个库。请确保项目中有 requirements.txt 文件，或直接安装这些库。主要依赖项包括：

   pip install python-dotenv langchain-mcp-adapters langgraph schedule langchain-core langchain-openai langchain-anthropic fastapi pydantic "uvicorn[standard]"
   

   上述命令包含核心依赖项，以及 FastAPI 服务和 CLI 工具所需的依赖项。文件 AgenticMaid/requirements.txt 列出了主要用于 API 和 CLI 功能的依赖项。

配置

AgenticMaid 支持多种配置方式：

1. Python 字典： 将 Python 字典直接传递给 AgenticMaid 构造函数。
2. JSON 文件： 向构造函数提供 JSON 配置文件的路径。
3. .env 文件： 对于默认值和 API 密钥等敏感信息，请在 AgenticMaid/ 目录（即与 client.py 同级）中创建 .env 文件。.env 文件中的值可由主 JSON/字典配置覆盖。

配置结构

主配置（Python 字典或 JSON）通常包含以下部分：

• model（可选）：AI 模型的全局默认设置。
• ai_services：各种 AI/大模型提供商和模型的定义。
• mcp_servers：客户端将连接的 MCP 服务配置。
• scheduled_tasks：按计划运行的任务数组。
• chat_services：客户端可与之交互的聊天服务定义。
• agents（可选）：预定义的 Agent 配置。
• default_llm_service_name（可选）：若未在其他位置指定，则使用的全局默认大模型服务。

详细示例及各字段的说明，请参阅 AgenticMaid/config.example.json 文件。

1. 使用 .env 文件

在 AgenticMaid 目录中创建名为 .env 的文件（例如 AgenticMaid/.env）。此文件用于存储 API 密钥和其他默认设置。其中的值将作为默认值，可由主配置文件或字典中的值覆盖。

AgenticMaid/.env 示例：

# AgenticMaid/.env example

# Default API key if not specified per service in main config
# DEFAULT_API_KEY=your_default_api_key_here

# Default model name if not specified per service in main config
# DEFAULT_MODEL_NAME=gpt-3.5-turbo-default-from-env

# Provider-specific defaults
OPENAI_API_KEY=your_openai_api_key_from_env
OPENAI_DEFAULT_MODEL=gpt-3.5-turbo-openai-from-env

ANTHROPIC_API_KEY=your_anthropic_api_key_from_env
ANTHROPIC_DEFAULT_MODEL=claude-2-from-env

# For Azure OpenAI
# AZURE_OPENAI_API_KEY=your_azure_openai_key
# AZURE_OPENAI_ENDPOINT=your_azure_endpoint
# AZURE_OPENAI_DEPLOYMENT_NAME=your_deployment_name

# For local OpenAI-compatible servers (e.g., LM Studio)
# LOCAL_OPENAI_BASE_URL=http://localhost:1234/v1

2. 主配置（JSON 或 Python 字典）

此配置用于定义 MCP 服务、AI 服务、任务和聊天界面的具体信息。

a. ai_services

定义你打算使用的每个 AI 服务。键是该服务的自定义名称。 字段包括 provider、model、api_key、base_url 等，参见 AgenticMaid/config.example.json。如果未在此处提供 api_key，则可以从 .env 文件中读取。

示例片段（来自 config.example.json）：

{
  "ai_services": {
    "openai_gemini_pro": {
      "provider": "Google",
      "model": "gemini-2.5-pro",
      "api_key": "your_google_api_key_here_or_leave_blank_to_use_env"
    },
    "anthropic_claude_opus": {
      "provider": "Anthropic",
      "model": "claude-4-opus"
    }
  }
}

b. mcp_servers

定义客户端应连接的 MCP 服务。MultiServerAgenticMaid（或相关类）将使用这些配置。 字段包括 adapter_type、base_url（用于 FastAPI）、command_template（用于 CLI）、name 和 description，参见 AgenticMaid/config.example.json。

示例片段（来自 config.example.json）：

{
  "mcp_servers": {
    "server_1_local_fastapi": {
      "adapter_type": "fastapi",
      "base_url": "http://localhost:8001/mcp/v1",
      "name": "Local FastAPI MCP Server"
    }
  }
}

c. scheduled_tasks

定义应按计划运行的任务。每个任务对象包括：

• name：任务的描述性名称。
• cron_expression：类似 cron 的表达式（目前支持简单形式，例如“每天 HH:MM”或“0 * * * *”，后者通过 schedule 库的解释方式表示每小时运行；如需支持完整 cron，可能需要在 _schedule_tasks 中进行自定义解析）。
• prompt：给 Agent 的指令或提示词。
• agent_id（可选）：引用 agents 部分中预定义的 Agent。
• model_config_name：此任务的 Agent 要使用的 AI 服务名称（来自 ai_services）。
• enabled：布尔值，true 表示启用任务，false 表示禁用任务。

示例片段（来自 config.example.json）：

{
  "scheduled_tasks": [
    {
      "name": "Hourly Summary Bot",
      "cron_expression": "0 * * * *", // Placeholder, actual parsing depends on _schedule_tasks
      "prompt": "Generate a brief summary of activities from the last hour.",
      "model_config_name": "openai_gemini_pro",
      "enabled": true
    }
  ]
}
}

cron_expression 的解释由 schedule 库处理。对于更复杂的 cron 字符串，可能需要调整 AgenticMaid/client.py 中 _schedule_tasks 的解析逻辑。

d. chat_services

定义不同聊天界面的配置。每个聊天服务对象包括：

• service_id：聊天服务的唯一标识符。
• llm_service_name：为此聊天提供支持的 AI 服务名称（来自 ai_services）。
• streaming_api_endpoint（概念性）：表示流式响应可能由何处提供的路径。
• non_streaming_api_endpoint（概念性）：用于非流式（完整）响应的路径。

示例片段（来自 config.example.json）：

{
  "chat_services": [
    {
      "service_id": "general_support_chat",
      "service_id": "general_support_chat",
      "llm_service_name": "openai_gemini_pro",
      "streaming_api_endpoint": "/chat/v1/streams/general_support_chat",
      "non_streaming_api_endpoint": "/chat/v1/completions/general_support_chat"
    }
  ]
}

双提示词系统

chat_services 配置现在支持双提示词系统，可为 Agent 的行为提供更多上下文和控制。这通过两个可选字段实现：system_prompt 和 role_prompt。

• system_prompt：此提示词会作为第一条 system 角色消息注入。它用于向 AI 模型提供适用于整个对话的高级指令、上下文或约束。例如，你可以定义 Agent 的角色设定、核心功能和操作边界。

• role_prompt：如果提供了系统提示词，此提示词会紧接其后、在实际用户消息之前作为一条 user 消息注入。它用于引导 AI 在接下来的对话轮次中应如何行动，或应采用什么特定角色。你可以借此设定特定语气，或引导 Agent 聚焦于当前任务。

处理聊天请求时，发送给 AI 模型的最终消息列表顺序如下：

1. 系统提示词（如果提供）
2. 角色提示词（如果提供）
3. 用户消息

config.json 中使用提示词的示例：

{
  "chat_services": [
    {
      "service_id": "general_support_chat_gemini",
      "llm_service_name": "google_gemini_default",
      "system_prompt": "You are a helpful and friendly customer support assistant for the AgenticMaid project. Your goal is to provide clear, accurate, and concise answers.",
      "role_prompt": "Please answer the user's question based on the project's documentation and capabilities. Be polite and professional.",
      "streaming_api_endpoint": "/chat/v1/streams/general_support_chat_gemini",
      "non_streaming_api_endpoint": "/chat/v1/completions/general_support_chat_gemini"
    }
  ]
}

用法

1. 初始化

首先，导入并初始化 AgenticMaid。创建实例后，需要调用 await client.async_initialize() 来完成异步设置（例如获取 MCP 工具）。

import asyncio
from pkg_AgenticMaid.client import ClientAgenticMaid # Placeholder: Actual class name from client.py

async def main():
    # Option 1: Load config from JSON file
    # client = ClientAgenticMaid(config_path_or_dict='AgenticMaid/config.example.json')

    # Option 2: Load config from a Python dictionary (Direct Python Invocation)
    # This method is ideal for embedding AgenticMaid within other Python applications,
    # allowing for dynamic configuration without relying on external JSON files.
    # The .env file for API keys and defaults is still loaded if present.
    config_dict = {
        "ai_services": {
            "my_gemini_service": { # Custom name for your service
                "provider": "Google",
                "model": "gemini-2.5-pro"
                # API key can be provided here directly: "api_key": "AIza...",
                # or if omitted, it will attempt to load from .env (e.g., GOOGLE_API_KEY)
            }
        },
        "mcp_servers": {
            "example_mcp_server": { # Custom name for your MCP server connection
                "adapter_type": "fastapi", # Or other supported adapter types
                "base_url": "http://localhost:8001/mcp/v1", # URL of the target MCP server
                "name": "My Example MCP Server"
            }
        },
        "default_llm_service_name": "my_gemini_service", # Default LLM for agents if not specified
        # Other sections like "scheduled_tasks", "chat_services", "agents" can be added as needed.
        # For a comprehensive, runnable example of direct dictionary invocation,
        # please refer to the script:
        # [`AgenticMaid/examples/direct_invocation_example.py`](./examples/direct_invocation_example.py)
    }
    client = ClientAgenticMaid(config_path_or_dict=config_dict)

    # Perform asynchronous initialization
    await client.async_initialize()

    if client.config and client.mcp_client:
        print("ClientAgenticMaid initialized successfully.")
        print(f"Fetched {len(client.mcp_tools)} MCP tools: {[tool.name for tool in client.mcp_tools]}")
    else:
        print("ClientAgenticMaid initialization failed or no MCP tools found.")
        print("AgenticMaidClient initialization failed or no MCP tools found.")
        return

    # ... use the client ...

if __name__ == "__main__":
    asyncio.run(main())

1.1. 直接字典调用的详细示例

如需查看可运行脚本，了解如何使用直接字典配置实例化并使用 ClientAgenticMaid，包括聊天等基本操作，请参阅示例文件：

• AgenticMaid/examples/direct_invocation_example.py

此示例展示了如何设置配置字典并执行常见的客户端操作。

2. 运行 MCP 交互（Agent 调用）

使用 run_mcp_interaction 方法与 Agent 交互。系统会使用指定的 LLM 服务和所有已获取的 MCP 工具创建 Agent（如果 Agent 已存在，则获取该 Agent）。

```python
# Assuming 'client' is an initialized ClientAgenticMaid instance from the example above

    # Example: Run an interaction
    messages_for_agent = [{"role": "user", "content": "What is the weather in London using available tools?"}]
    llm_service_to_use = "my_gemini_service" # Must be a key from your ai_services config
    agent_identifier = "weather_agent_01" # A custom key for this agent instance

    response = await client.run_mcp_interaction(
        messages=messages_for_agent,
        llm_service_name=llm_service_to_use,
        agent_key=agent_identifier
    )

    if response and "error" not in response:
        print(f"Agent Response: {response}")
    else:
        print(f"Agent Interaction Error: {response.get('error') if response else 'Unknown error'}")

3. 运行计划任务

要运行计划任务，请先确保已在配置中定义这些任务，然后启动调度器。调度器会在后台线程中运行。

# Assuming 'client' is an initialized ClientAgenticMaid instance

    # To start the scheduler (it runs in a background thread):
    if client.scheduler.jobs: # Check if there are any jobs scheduled
        print("Starting scheduler...")
        client.start_scheduler()
        # The scheduler will now run tasks in the background.
        # Keep the main thread alive if you want tasks to continue running.
        # For example, in a long-running application:
        # try:
        #     while True:
        #         await asyncio.sleep(1)
        # except KeyboardInterrupt:
        #     print("Application shutting down.")
        #     client.stop_scheduler() # Conceptual stop
    else:
        print("No tasks scheduled.")

注意： start_scheduler 方法会在守护线程中运行无限循环。请确保主应用妥善管理其生命周期。stop_scheduler 方法目前只是占位符；在复杂应用中，可能需要更可靠的停止机制（例如使用 threading.Event）以实现平稳关闭。

4. 与聊天服务交互

要处理已定义聊天服务的消息，请使用 handle_chat_message 方法。

# Assuming 'client' is an initialized ClientAgenticMaid instance

    # Example: Interact with a chat service
    chat_service_id_to_use = "general_support_chat" # Must be a service_id from your chat_services config
    chat_messages = [{"role": "user", "content": "Hello, I need help with my account."}]

    chat_response = await client.handle_chat_message(
        service_id=chat_service_id_to_use,
        messages=chat_messages,
        stream=False # Set to True for streaming (currently placeholder)
    )

    if chat_response and "error" not in chat_response:
        print(f"Chat Service Response: {chat_response}")
    else:
        print(f"Chat Service Error: {chat_response.get('error') if chat_response else 'Unknown error'}")

7. 多 Agent 调度

AgenticMaid 支持多 Agent 调度功能，允许一个 Agent 调用另一个 Agent。借助此功能，可以创建复杂的分层 Agent 结构，让主 Agent 将特定任务委派给专用 Agent。

a. 配置

要启用此功能，必须在 config.json 文件中添加 multi_agent_dispatch 部分。

配置字段：

• enabled（布尔值）：设为 true 以启用此功能。
• default_mode（字符串）：确定默认调用模式。
  • synchronous 或 sync：调用方 Agent 会等待目标 Agent 完成任务并返回结果。
  • concurrent：调用方 Agent 会调用目标 Agent，并立即继续执行自身任务，不等待结果。
• allowed_invocations（对象）：定义允许哪些 Agent 调用其他 Agent 的字典。
  • 键为配置中 agents 部分的调用方 Agent 的 agent_id。
  • 值为字符串数组，每个字符串都是可调用的目标 Agent 的 agent_id。
  • 可在数组中使用通配符 "*"，允许某个 Agent 调用其他所有已配置的 Agent。

config.json 片段示例：

{
  "multi_agent_dispatch": {
    "enabled": true,
    "default_mode": "concurrent",
    "allowed_invocations": {
      "orchestrator_agent": [
        "*"
      ],
      "summary_agent_config_ref": [
        "report_agent_v2"
      ],
      "report_agent_v2": []
    }
  },
  "agents": {
    "orchestrator_agent": {
      "model_config_name": "google_gemini_default"
    },
    "summary_agent_config_ref": {
      "model_config_name": "google_gemini_default"
    },
    "report_agent_v2": {
      "model_config_name": "anthropic_claude4_opus"
    }
  }
}

在此示例中：

• orchestrator_agent 可以调用任何其他 Agent。
• summary_agent_config_ref 只能调用 report_agent_v2。
• report_agent_v2 不能调用其他任何 Agent。

b. 在提示词中使用

启用此功能后，系统会自动向获准调用其他 Agent 的 Agent 提供 dispatch 工具。要使用该工具，请在提示词中指示 Agent 使用所需参数调用 dispatch 工具。

Dispatch 工具参数：

• agent_id（字符串）：要调用的目标 Agent 的 ID。
• prompt（字符串）：要传递给目标 Agent 的提示词或指令。
• mode（字符串，可选）：调用模式（sync 或 concurrent）。如果省略，则使用配置中的 default_mode。

提示词示例：

"Please use the dispatch tool to ask the 'report_agent_v2' to generate a detailed analysis of the latest user feedback. Run this in sync mode."

随后，Agent 会解析此指令并执行以下工具调用：dispatch(agent_id='report_agent_v2', prompt='Generate a detailed analysis of the latest user feedback.', mode='sync')。

示例

完整示例脚本（example_usage.py）

import asyncio
import time
from pkg_AgenticMaid.client import ClientAgenticMaid # Placeholder: Actual class name from client.py

async def run_client_operations():
    config = {
        "ai_services": {
            "default_llm": {
                "provider": "Google", # Ensure GOOGLE_API_KEY is in .env
                "model": "gemini-2.5-pro"
            },
            "claude_opus_llm": {
                 "provider": "Anthropic", # Ensure ANTHROPIC_API_KEY is in .env
                 "model": "claude-4-opus"
            }
        },
        "mcp_servers": {
            # Define at least one MCP server for tools to be fetched.
            # This example assumes an MCP server is running at http://localhost:8001/mcp/v1
            # If not, mcp_tools will be empty.
            "my_mcp_server": {
                "adapter_type": "fastapi",
                "base_url": "http://localhost:8001/mcp/v1", # Replace with your actual MCP server URL
                "name": "Example MCP Server"
            }
        },
        "scheduled_tasks": [
            {
                "name": "Test Scheduled Task",
                "cron_expression": "daily at 00:00", # Will run once if current time is past 00:00 and scheduler is kept running
                "prompt": "This is a test scheduled prompt. What time is it using Gemini?",
                "model_config_name": "default_llm",
                "enabled": True # Set to False if you don't want it to run
            }
        ],
        "chat_services": [
            {
                "service_id": "test_chat_gemini",
                "llm_service_name": "default_llm"
            },
            {
                "service_id": "test_chat_claude",
                "llm_service_name": "claude_opus_llm"
            }
        ],
        "default_llm_service_name": "default_llm"
    }

    client = ClientAgenticMaid(config_path_or_dict=config)
    await client.async_initialize()

    if not client.config:
        print("Client configuration failed. Exiting.")
        return

    print(f"ClientAgenticMaid Initialized. Config Source: {client.config_source}")
    print(f"Available MCP Tools: {[tool.name for tool in client.mcp_tools] if client.mcp_tools else 'No tools fetched (check MCP server config and availability)'}")

    # 1. Agent Interaction with Gemini
    print("\n--- Testing Agent Interaction (Gemini) ---")
    interaction_messages_gemini = [{"role": "user", "content": "Tell me a fun fact using Gemini."}]
    interaction_response_gemini = await client.run_mcp_interaction(
        messages=interaction_messages_gemini,
        llm_service_name="default_llm", # Uses gemini-2.5-pro
        agent_key="fun_fact_agent_gemini"
    )
    print(f"Agent Interaction Response (Gemini): {interaction_response_gemini}")

    # 1b. Agent Interaction with Claude
    print("\n--- Testing Agent Interaction (Claude) ---")
    interaction_messages_claude = [{"role": "user", "content": "Tell me a different fun fact using Claude."}]
    interaction_response_claude = await client.run_mcp_interaction(
        messages=interaction_messages_claude,
        llm_service_name="claude_opus_llm", # Uses claude-4-opus
        agent_key="fun_fact_agent_claude"
    )
    print(f"Agent Interaction Response (Claude): {interaction_response_claude}")


    # 2. Chat Service with Gemini
    print("\n--- Testing Chat Service (Gemini) ---")
    chat_messages_gemini = [{"role": "user", "content": "Hi there, how are you? (Gemini)"}]
    chat_response_gemini = await client.handle_chat_message(
        service_id="test_chat_gemini",
        messages=chat_messages_gemini
    )
    print(f"Chat Service Response (Gemini): {chat_response_gemini}")

    # 2b. Chat Service with Claude
    print("\n--- Testing Chat Service (Claude) ---")
    chat_messages_claude = [{"role": "user", "content": "Hi there, how are you? (Claude)"}]
    chat_response_claude = await client.handle_chat_message(
        service_id="test_chat_claude",
        messages=chat_messages_claude
    )
    print(f"Chat Service Response (Claude): {chat_response_claude}")

    # 3. Scheduled Tasks
    print("\n--- Testing Scheduled Tasks ---")
    if client.scheduler.jobs:
        print(f"Scheduled jobs: {client.scheduler.jobs}")
        print("Starting scheduler for a short period (e.g., 5 seconds for demo)...")
        client.start_scheduler() # Starts a daemon thread

        # Keep the main script running for a bit to allow scheduler to work
        # In a real app, this would be part of the main application loop.
        # For this demo, we'll just sleep.
        # Note: 'daily at HH:MM' tasks might not run in this short window unless HH:MM is very soon.
        # Consider a more frequent cron_expression for immediate testing, e.g., using a custom parser for 'every X seconds'.
        await asyncio.sleep(5)
        print("Scheduler demo period finished.")
        # client.stop_scheduler() # Conceptual
    else:
        print("No tasks scheduled.")

if __name__ == "__main__":
    # Note: If your MCP server or .env setup is not complete, parts of this example might show warnings or errors.
    # Ensure an MCP server is running if you expect tools, and .env has API keys for LLM calls.
    print("Make sure your .env file (in AgenticMaid directory) has GOOGLE_API_KEY and ANTHROPIC_API_KEY set for this example to fully work.")
    print("Also, ensure an MCP server is running at the configured URL if you expect MCP tools.")
    asyncio.run(run_client_operations())

本 README 提供了安装、配置和使用 ClientAgenticMaid 的全面指南。请记得根据项目的具体设置调整路径和配置。