Auto Maid

AgenticMaid项目

概述

AgentMaid是一个Python库，旨在与一个或多个多功能协议（MCP）服务器交互。它允许动态获取和利用这些服务器提供的工具（功能）。客户端还可以管理各种AI/LLM服务的配置，安排自动化任务，处理聊天服务交互，使其成为构建AI驱动应用程序的通用组件。

它利用 langchain-mcp-adapters 用于与MCP服务器通信 langgraph 用于创建可以使用提取的MCP工具的反应剂。

特性

• 多服务器MCP交互： 连接到多个MCP服务器并利用其中的工具。
• 动态刀具抓取： 在运行时从MCP服务器检索可用工具。
• 灵活配置： 支持通过Python字典、JSON文件和 .env 敏感数据的文件。
• 人工智能服务管理： 配置和利用各种AI/LLM服务（例如OpenAI、Anthropic、Azure OpenAI、本地模型）。
• 计划任务： 允许根据类似cron的计划定义和运行任务。
• 聊天服务集成： 提供一个框架，用于处理与已定义聊天服务的交互。
• 代理创建： 用途 langgraph 创建可以利用MCP工具和配置LLM的ReAct代理。
• 环境变量支持： 从 .env 文件。

安装

1. 先决条件：

- Python 3.8+

1. 克隆存储库（如果适用）或添加 AgenticMaid 到你的项目。

1. 安装依赖关系：

客户端依赖于多个库。确保你有一个 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密钥），创建 .env 文件在 AgenticMaid/ 目录（即，旁边 client.py).价值观来自 .env 文件可以被主JSON/字典配置覆盖。

配置结构

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

• model （可选）：AI模型的全局默认设置。
• ai_services：各种人工智能/法学硕士提供者和模型的定义。
• mcp_servers：客户端将连接到的MCP服务器的配置。
• scheduled_tasks：按计划运行的一系列任务。
• chat_services：客户端可以与之交互的聊天服务的定义。
• agents （可选）：预定义的代理配置。
• default_llm_service_name （可选）：如果未在其他地方指定，则使用全局默认LLM服务。

请参阅 AgenticMaid/config.example.json 文件中提供了一个详细的示例，并附有解释每个字段的注释。

1.使用 .env 文件

创建一个名为的文件 .env 在 AgenticMaid 目录（例如。， 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 Dict）

此配置定义了MCP服务器、AI服务、任务和聊天界面的具体细节。

一 ai_services

定义您打算使用的每个AI服务。密钥是服务的自定义名称。 参见 AgenticMaid/config.example.json 对于以下领域 provider, model, api_key, base_url等等。这 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 （或相关类）将使用这些配置。 参见 AgenticMaid/config.example.json 对于以下领域 adapter_type, base_url （适用于FastAPI）， command_template （对于CLI）， name，以及 description.

示例片段（来自 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的表达式（目前支持简单的形式，如“day at HH:MM”或“0\*\*\*\*”表示小时，通过 schedule 库的解释，这可能需要在中进行自定义解析 _schedule_tasks 对于完整的cron）。
• prompt：代理人的指示/提示。
• agent_id （可选）：在 agents 部分。
• model_config_name：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字符串 _schedule_tasks 在...之内 AgenticMaid/client.py 可能需要调整。

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 配置现在支持双提示系统，以提供对代理行为的更多上下文和控制。这是通过两个可选字段实现的： system_prompt 和 role_prompt.

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

• role_prompt：此提示作为 user 在系统提示（如果提供）之后和实际用户消息之前立即发送消息。它用于指导人工智能在接下来的对话中应该如何表现，或者应该扮演什么具体角色。这对于设定特定基调或引导代理专注于当前任务非常有用。

当处理聊天请求时，发送到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交互（代理调用）

使用 run_mcp_interaction 与代理交互的方法。将使用指定的LLM服务和所有提取的MCP工具创建（或检索（如果已存在）代理。

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.多代理调度

这 `AgenticMaid` 支持多代理调度功能，允许一个代理调用另一个代理。这使得能够创建复杂的分层代理结构，其中主代理可以将特定任务委托给专门的代理。

#### a.配置

要启用此功能，您必须添加 `multi_agent_dispatch` 部分到您的 `config.json` 文件。

**配置字段：**

- `enabled` （boolean）：设置为 `true` 以启用该功能。
- `default_mode` （string）：确定默认调用模式。
  - `synchronous` 或 `sync`：调用代理等待目标代理完成其任务并返回结果。
  - `concurrent`：调用代理调用目标代理，并立即继续自己的执行，而无需等待结果。
- `allowed_invocations` （object）：定义哪些代理可以呼叫其他代理的字典。
  - 钥匙是 `agent_id` 的 *召唤* 代理人（来自 `agents` 配置的一部分）。
  - 这些值是一个字符串数组，其中每个字符串都是 `agent_id` a *目标* 可以调用的代理。
  - 通配符 `"*"` 可以在数组中使用，以允许代理调用 *任何* 其他配置的代理。

**示例 `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` 可以呼叫任何其他代理。
- `summary_agent_config_ref` 只能呼叫 `report_agent_v2`.
- `report_agent_v2` 无法呼叫任何其他代理。

#### b.提示中的用法

启用该功能后 `dispatch` 该工具会自动提供给被允许呼叫其他人的代理。要使用它，请在提示中指示代理调用 `dispatch` 工具具有所需的参数。

**调度工具参数：**

- `agent_id` （string）：要调用的目标代理的ID。
- `prompt` （string）：传递给目标代理的提示或指令。
- `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."

然后，代理将解析此指令并执行以下工具调用： `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`。请记住根据您的特定项目设置调整路径和配置。