AgenticMaid Project

AgenticMaid プロジェクト

概要

AgenticMaid は、1 つ以上の Multi-Capability Protocol（MCP）サーバーと連携するために設計された Python ライブラリです。これらのサーバーが提供するツール（機能）を動的に取得して利用できます。また、クライアントはさまざまな AI／LLM サービスの設定管理、自動タスクのスケジュール設定、チャットサービスとのやり取りにも対応しているため、AI を活用したアプリケーションを構築するための汎用性の高いコンポーネントとなっています。

MCP サーバーとの通信には langchain-mcp-adapters を、取得した MCP ツールを利用できるリアクティブエージェントの作成には langgraph を活用しています。

機能

• 複数サーバーとの MCP 連携： 複数の MCP サーバーに接続し、それらのツールを利用します。
• 動的なツール取得： MCP サーバーから利用可能なツールを実行時に取得します。
• 柔軟な設定： Python 辞書、JSON ファイル、機密データ用の .env ファイルによる設定に対応しています。
• AI サービス管理： さまざまな AI／LLM サービス（OpenAI、Anthropic、Azure OpenAI、ローカルモデルなど）を設定して利用します。
• スケジュールタスク： cron に似たスケジュールに基づいてタスクを定義し、実行できます。
• チャットサービスとの連携： 定義されたチャットサービスとのやり取りを処理するフレームワークを提供します。
• エージェントの作成： langgraph を使用して、MCP ツールと設定済みの LLM を活用できる ReAct エージェントを作成します。
• 環境変数のサポート： .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／LLM プロバイダーとモデルの定義。
• mcp_servers：クライアントが接続する MCP サーバーの設定。
• scheduled_tasks：スケジュールに従って実行するタスクの配列。
• chat_services：クライアントがやり取りできるチャットサービスの定義。
• agents（任意）：事前定義されたエージェントの設定。
• default_llm_service_name（任意）：ほかで指定されていない場合に使用する、グローバルなデフォルト LLM サービス。

各フィールドを説明するコメントを含む詳細な例については、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 Dict）

この設定では、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 形式の式（現在は「daily at HH:MM」や、schedule ライブラリの解釈による1時間ごとの実行を表す「0 * * * *」のような簡単な形式をサポートしています。完全な cron 形式に対応するには、_schedule_tasks に独自の解析処理が必要になる場合があります）。
• 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 文字列を使用する場合は、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 設定では、エージェントの動作に関するコンテキストと制御をさらに提供するデュアルプロンプトシステムをサポートするようになりました。これは、system_prompt と role_prompt の2つの任意フィールドで実現します。

• system_prompt：system ロールの最初のメッセージとしてこのプロンプトが挿入されます。会話全体に適用される高度な指示、コンテキスト、制約を AI モデルに与えるために使用します。たとえば、エージェントのペルソナ、中心的な機能、動作上の境界を定義できます。

• role_prompt：system_prompt が指定されている場合はその直後、実際のユーザーメッセージの前に、user メッセージとしてこのプロンプトが挿入されます。会話の次のターンでのエージェントの振る舞いや、担うべき特定の役割を AI に示すために使用します。特定のトーンを設定したり、直近のタスクに対するエージェントの重点を指定したりする場合に便利です。

チャットリクエストを処理すると、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 をインポートして初期化します。非同期セットアップ（MCP ツールの取得など）を完了するには、インスタンスを作成した後に await client.async_initialize() を呼び出す必要があります。

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. 設定

この機能を有効にするには、config.json ファイルに multi_agent_dispatch セクションを追加する必要があります。

設定フィールド：

• enabled（ブール値）：機能を有効にするには true に設定します。
• default_mode（文字列）：デフォルトの呼び出しモードを指定します。
  • synchronous または sync：呼び出し元のエージェントは、対象エージェントがタスクを完了して結果を返すまで待機します。
  • concurrent：呼び出し元のエージェントは対象エージェントを呼び出した後、結果を待たずに自身の実行を続けます。
• allowed_invocations（オブジェクト）：どのエージェントがほかのエージェントを呼び出せるかを定義する辞書です。
  • キーは、設定の agents セクションにある、呼び出し元エージェントの agent_id です。
  • 値は文字列の配列で、各文字列は呼び出し可能な対象エージェントの agent_id です。
  • 配列内でワイルドカード "*" を使用すると、そのエージェントから設定済みの任意のエージェントを呼び出せるようになります。

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（文字列）：呼び出す対象エージェントの ID です。
• prompt（文字列）：対象エージェントに渡すプロンプトまたは指示です。
• 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 のインストール、設定、使用方法を詳しく説明しています。パスと設定は、使用するプロジェクトのセットアップに合わせて調整してください。