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/v12.主配置（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 使用直接字典配置，包括聊天等基本操作，请参阅示例文件：
此示例展示了如何设置配置字典并执行常见的客户端操作。
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`。请记住根据您的特定项目设置调整路径和配置。