メインコンテンツまでスキップ

はじめてのエージェント

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]==3.0.15

ステップ 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]==3.0.15'
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 する形でも二重送信にはなりません
  • サイクル型の実行（mp-chat/1。プロジェクトのサイクルが周期や Webhook で起動する会話）では、上の変数に加えて PROJECT_ID（そのプロジェクトの ID）が注入されます。プロジェクトの連携先のうち有効なものは、OAuth のトークンが ASTER_CONNECTION_<SERVICE>_TOKEN（<SERVICE> は接続先の service 名を大文字にし、- を _ にしたもの。例: slack_mcp → ASTER_CONNECTION_SLACK_MCP_TOKEN）で渡ります。有効な連携先が 8 件を超えるとき、同じ service の連携先が 2 つ有効なとき、この変数名がエージェント登録や配置の環境変数とぶつかるときは、実行は起動しません。接続が切れている接続先や、期限が過ぎていて更新できない接続先のトークンは渡らないので（期限が過ぎていても、更新できれば渡ります）、変数が無いときの動きも作ってください。接続先の URL は、エージェントが自分で設定（get_mcp_configurations(execution_mode="pod") など）から引きます
  • サイクル型の実行（mp-chat/1。プラットフォームが予定で起動する会話）では、v3.0.12 から runner が agent を呼ぶ前に開始済みの印を DB に取ります。Pod が作り直されて同じ実行がもう一度起動しても agent は呼ばれず、MarketplaceRunnerFenceTaken で終わります（プラットフォームが効果不明の失敗として閉じます）。印には DB 直結のプール（PostgreSQLManager.pool）を使うので、プールの無い data_access では started_fence=False を渡してください（その場合、作り直された Pod で agent がもう一度動くことがあります）

サイクル型の実行はプロセスを終了させる​

自分でサーバを立てる場合は必ず対応してください

サイクル型の実行（mp-chat/1）は、1 回の実行ごとに Pod が作られます（restartPolicy: Never）。 プラットフォームはコンテナの終了コードで実行の終わりを判定するため、処理が終わってもプロセスが 残っていると、その実行は終わったことになりません。

終わらないままだと、その配置の次の発火は見送られ続けます（前の実行がまだ動いている扱いになる）。

サイクル型は会話型と同じ protocol で動くので、ヘルスチェックに応答し続けるサーバの例をそのまま 使うと起きます。

run_marketplace_agent に渡す形（上記）なら、runner が終端まで面倒を見るので追加の対応は要りません。

自分で FastAPI / uvicorn などのサーバを立てる場合は、処理の完了後にサーバを閉じて終了します。 処理中はヘルスチェックに応答させたいので、閉じるのは完了を確認してからにします。

start_api_server.py（サンプルエージェントの実装）Python
import asyncio, os
import uvicorn

# 実行の完了を知らせるフラグ（エージェント本体の最後で set する）
_run_finished = asyncio.Event()


def exit_after_run() -> bool:
    """サイクル型は 1 回で終わる = 実行の完了でプロセスを終える。"""
    return os.environ.get("AGENT_MODE", "").strip().lower() == "cycle"


async def serve(config: uvicorn.Config) -> None:
    server = uvicorn.Server(config)
    serving = asyncio.create_task(server.serve())
    if exit_after_run():
        finished = asyncio.create_task(_run_finished.wait())
        # サーバが先に終わったとき（起動失敗・SIGTERM）に完了を待ち続けない
        await asyncio.wait({serving, finished}, return_when=asyncio.FIRST_COMPLETED)
        finished.cancel()
        # 立ち上がり切る前に閉じると後片付けが打ち切られるので、started を待つ
        for _ in range(100):  # 上限 5 秒
            if server.started or serving.done():
                break
            await asyncio.sleep(0.05)
        server.should_exit = True
    await serving


if __name__ == "__main__":
    asyncio.run(serve(uvicorn.Config("main:app", host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))))
  • サイクル型かどうかは、エージェント登録の環境変数（例: AGENT_MODE=cycle）で切り替えます。会話型と 同じイメージを使う場合は、会話型では常駐のままにします
  • ループの用意に関わる uvicorn の API（Config.setup_event_loop / Config.get_loop_factory）は 呼ばないでください。版によって有無が変わり、setup_event_loop は 0.36.0 で廃止されています
  • 例外で終わる場合も終了コードは 0 にします（runner を使えばそうなります）。異常の内容は 完遂の報告で伝えてください

サイクル型の実行で完遂を報告する（v3.0.14+）​

サイクル型の実行は、agent が正常に終わっても「実行終了（完遂は未計測）」として記録され、成功の件数に入りません。プロセスの正常終了を完遂とは見なさないためです。成功として数えるには、agent が目的を果たしたかを report_completion で報告します。報告は任意で、報告しない agent の挙動は変わりません。

marketplace_agent.py（完遂を報告する）Python
import asyncio
import os

from agenticstar_platform import arun_marketplace_agent, report_completion
from agenticstar_platform.db import DataAccess, PostgreSQLConfig, PostgreSQLManager


async def main() -> None:
    # runner と同じ接続を使うため、DataAccess を自分で作って runner に渡す
    manager = PostgreSQLManager(PostgreSQLConfig.from_env())
    async with DataAccess(manager) as data_access:

        async def my_agent(emitter, message: str) -> str:
            reply = message.upper()  # ここを実際の処理に置き換える
            await report_completion(
                manager.pool,
                user_id=os.environ["USER_ID"],
                conversation_id=os.environ["CONVERSATION_ID"],
                message_id=os.environ["MESSAGE_ID"],
                execution_id=os.environ["EXECUTION_ID"],
                outcome="completed",          # 目的を果たせなかったなら "incomplete" など
                had_mutating_success=False,   # 外部へ書き込んだか（分からなければ None）
            )
            return reply

        await arun_marketplace_agent(my_agent, data_access=data_access)


if __name__ == "__main__":
    asyncio.run(main())
  • outcome は completed / incomplete / failed / timeout / stopped / denied / skipped から選びます。何をもって完遂とするかは agent が決めます
  • 理由を付けるときは reason に短い snake_case のコード（例: no_new_items）を渡します。文章は summary に入れてください。プラットフォームが実行の結果に使うのは outcome と reason だけです
  • 報告は 1 回だけです（2 回目は、同時に呼んだ場合も書かれず、False が返ります）。DB への書き込みに失敗しても例外にはならず、その実行は未計測のままです
  • プラットフォームが報告を採用するのは、コンテナが exit 0 で終わったのを確かめたときだけです（取消・agent の実行中の期限超過・Pod の消失では採用されません）。agent が例外で終わっても runner はプロセスを exit 0 で終えるため、報告済みならその報告が採用されます。報告は結果が確定してから（agent 関数の最後で）呼んでください
  • 通常の会話（チャット）で呼んでも結果は変わりません。引数と動作の詳細は SDK API リファレンス を参照してください

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]==3.0.15'
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

次のステップ​