はじめてのエージェント

agenticstar-platform SDK を使って、最小構成のカスタムエージェントを作成・実行するまでの手順です。SDK はインフラ（DB・イベント・ストレージ等）を提供し、エージェントロジックは開発者が自由に設計します。

前提条件

• Python 3.12 以上
• pip または uv
• PostgreSQL データベース（ローカルまたは Azure Database for PostgreSQL）
• Azure OpenAI API キー（LLM 呼び出し用）

ステップ 1: SDK のインストール

全モジュール

curl

pip install agenticstar-platform[all]==0.5.40

最小構成

curl

# イベント + DB のみ（最小構成）
pip install agenticstar-platform[db]==0.5.40

uv を使う場合

curl

uv pip install agenticstar-platform[all]==0.5.40

ステップ 2: 設定ファイルの作成

config.toml を作成します。SDK の各モジュールはこの設定ファイルから構成を読み取ります。

config.tomlToml

[database]
host = "localhost"
port = 5432
database = "agenticstar"
username = "agent_user"
password = "your_password"
pool_min_size = 2
pool_max_size = 10
# Azure AD 認証を使う場合のみ true にし、[database.azure_ad] を設定
use_azure_ad = false

[events]
enable_db_handler = true
enable_webhook_handler = false

PostgreSQLConfig.from_toml() は既定で [database] セクションを読み込みます。フィールド名は username / pool_min_size / pool_max_size です。Azure AD 認証を使う場合は [database] 直下に use_azure_ad = true を置き、[database.azure_ad] に tenant_id / client_id / client_secret を設定します。

ステップ 3: エージェントの実装

my_agent.py を作成します。この例では、SDK の DB モジュール でジョブ情報を取得し、Events モジュール で進捗をストリーミング配信するシンプルなエージェントを実装します。

my_agent.pyPython

import asyncio
from openai import AsyncAzureOpenAI

from agenticstar_platform.db import PostgreSQLManager, PostgreSQLConfig, DataAccess
from agenticstar_platform.db.execution_access import ExecutionAccess
from agenticstar_platform.events import EventEmitter, EventType

# --- 設定 ---
db_config = PostgreSQLConfig.from_toml("config.toml")
llm_client = AsyncAzureOpenAI(
    azure_endpoint="https://your-endpoint.openai.azure.com/",
    api_version="2024-12-01-preview",
    api_key="your-api-key",
)

async def run_agent(execution_id: str, user_message: str):
    """最小構成のエージェント実行フロー"""

    # 1. DB 接続を初期化
    db_manager = PostgreSQLManager(db_config)
    da = DataAccess(db_manager)
    await da.initialize()

    execution_access = ExecutionAccess(da)

    # 2. イベントエミッター初期化
    emitter = EventEmitter(execution_id=execution_id)

    # 3. 実行開始イベント
    await emitter.emit_event(
        event_type=EventType.PHASE_START,
        message="リクエストを処理中...",
        metadata={"phase": "processing"},
    )

    # 4. 過去のメッセージを取得
    history = await execution_access.get_messages(execution_id=execution_id)

    # 5. LLM 呼び出し（エージェントロジックは自由に設計）
    messages = [{"role": "system", "content": "あなたは親切なアシスタントです。"}]
    if history:
        for h in history:
            messages.append({"role": h["role"], "content": h["content"]})
    messages.append({"role": "user", "content": user_message})

    response = await llm_client.chat.completions.create(
        model="gpt-4.1",
        messages=messages,
    )
    answer = response.choices[0].message.content

    # 6. 完了イベント
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message=answer,
    )

    # 7. クリーンアップ
    await emitter.cleanup()
    await da.close()

    return answer

if __name__ == "__main__":
    result = asyncio.run(run_agent("exec-001", "こんにちは！"))
    print(result)

ステップ 4: 動作確認

curl

python my_agent.py

正常に動作すると、LLM のレスポンスがコンソールに出力されます。

Marketplace 互換で実行する（runner・SDK 0.5.30+）

ローカルで動いた agent 関数は、run_marketplace_agent に渡すだけで Marketplace 互換の終端ライフサイクルで実行できます。identity（EXECUTION_ID 等）の受領・検証、 入力メッセージの取得、結果の DB 保存、Webhook 通知、終端イベント（何が起きても 正確に 1 回）、cleanup は runner が担うため、上のステップ 3 で書いたような main ボイラープレートは不要になります。

marketplace_agent.pyPython

from agenticstar_platform import run_marketplace_agent


async def my_agent(emitter, message: str) -> str:
    """エージェント本体。ロジックは自由（LangChain / OpenAI Agents SDK / 自作）。"""
    return message.upper()  # ここを実際の処理に置き換える


if __name__ == "__main__":
    run_marketplace_agent(my_agent)

curl

pip install 'agenticstar-platform[runner]==0.5.40'
python marketplace_agent.py

プラットフォーム上では EXECUTION_ID / CONVERSATION_ID / USER_ID / MESSAGE_ID は Marketplace executor が Pod 起動時に注入します。DB 接続 （DB_HOST 等、PostgreSQLConfig.from_env() 契約）と WEBHOOK_URL は エージェント登録時の環境変数として設定します。必須変数が欠けている場合、 runner は agent を呼ばずに MarketplaceRunnerConfigError で停止します。

• agent 関数の戻り値が completion_success として保存・通知されます
• agent 関数の例外は completion_failure に収束します（traceback はログのみ）
• agent 関数が自分で終端イベントを emit する形でも二重送信にはなりません

Local Integration Lab（SDK 0.5.35+）

外部サービスの契約・cloud credential・.env 編集なしで、DB / RAG / storage を跨ぐ 統合 journey（ingest → retrieve → artifact/result persist → terminal outcome）を 1 コマンドで体験できます。必要なのは Docker（compose v2 付き）と Python 3.11+ だけです。

curl

pip install 'agenticstar-platform[lab]==0.5.40'
python -m agenticstar_platform.lab            # journey 実行（1 コマンド）
python -m agenticstar_platform.lab doctor     # read-only 診断
python -m agenticstar_platform.lab --reset    # 破棄（対象を列挙して確認 / --yes で省略）

• PostgreSQL / Qdrant / S3 互換 storage (MinIO) を version 固定の Docker Compose で 自動起動します（127.0.0.1 のみに bind）
• embedding は決定論的な offline 実装で、cloud endpoint / API key を一切使いません
• 使用される credential は lab 専用の非本番固定値のみで、production credential を 要求・保存・表示しません
• artifact のローカルコピーは実行時のカレントディレクトリの agenticstar-lab-artifacts/ に置かれます
• 成功時に retrieved document・artifact の保存先・次に試す Marketplace runner の コマンドが表示されます。うまく動かないときは doctor が原因（Docker 未起動 / port 競合 / extra 不足など）と次の 1 コマンドを提示します

ステップ 5: ツールの追加（オプション）

エージェントにツール（関数呼び出し）を追加する例です。SDK の RAG モジュール を使ってナレッジベースを検索します。

tools/search_knowledge.pyPython

from agenticstar_platform.rag import QdrantManager, QdrantConfig
from agenticstar_platform.rag import EmbeddingGenerator, EmbeddingConfig

async def search_knowledge(query: str, top_k: int = 5) -> list[dict]:
    """ナレッジベースからクエリに関連するドキュメントを検索"""

    # Embedding 生成器と Qdrant マネージャを初期化
    # QdrantManager は EmbeddingGenerator を必須引数に取る（検索時に内部で埋め込みを生成）
    embed_config = EmbeddingConfig.from_toml("config.toml")
    generator = EmbeddingGenerator(embed_config)

    qdrant_config = QdrantConfig.from_toml("config.toml")  # collection_name は設定から
    qdrant = QdrantManager(qdrant_config, generator)
    await qdrant.initialize()

    # クエリテキストで類似検索（query_vector は不要。埋め込みは内部生成）
    result = await qdrant.search(query_text=query, limit=top_k)

    # 戻り値は dict。ヒットは result["data"]["results"] に格納される
    return [
        {"content": hit["payload"].get("content", ""), "score": hit["score"]}
        for hit in result["data"]["results"]
    ]

プロジェクト構成の例

my-agent/
├── config.toml              # SDK 設定ファイル
├── my_agent.py              # エージェントのメインロジック
├── tools/
│   ├── search_knowledge.py  # RAG 検索ツール
│   └── file_manager.py      # ストレージ操作ツール
├── Dockerfile               # コンテナビルド用
├── requirements.txt
└── README.md

次のステップ

• アーキテクチャガイド — SDK のモジュール構成と設計思想
• デプロイガイド — Docker ビルドとプラットフォームへの登録
• イベント / ストリーミングガイド — SSE イベント配信の詳細
• SDK API リファレンス — 全モジュール・クラスの仕様