使用说明

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 文件。

安装

先决条件：
- Python 3.8+
克隆存储库（如果适用）或添加 AgenticMaid 到你的项目。
安装依赖关系： 客户端依赖于多个库。确保你有一个 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 可以通过多种方式进行配置：

Python词典： 将Python字典直接传递给 AgenticMaid 建设者。
JSON文件： 为构造函数提供JSON配置文件的路径。
.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模型的最终消息列表将按以下顺序排列：

系统提示（如果提供）
角色提示（如果提供）
用户消息

带有提示的示例 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工具创建（或检索（如果已存在）代理。

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

这 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。请记住根据您的特定项目设置调整路径和配置。

使用说明

AgenticMaid项目

概述

特性

安装

配置

配置结构

1.使用 .env 文件

2.主配置（JSON或Python Dict）

一 ai_services

b mcp_servers

c scheduled_tasks

d chat_services

双提示系统

用法

1.初始化

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

2.运行MCP交互（代理调用）

3.运行计划任务

4.与聊天服务互动

7.多代理调度

a.配置

b.提示中的用法

例子

完整示例脚本(example_usage.py)

相关 MCP

文件转Markdown工具

MCP服务器交互工具

架构图生成服务

零依赖MCP服务器

MCP命令行工具

MCP服务器开发库

1.使用 `.env` 文件

一 `ai_services`

b `mcp_servers`

c `scheduled_tasks`

d `chat_services`

完整示例脚本(`example_usage.py`)