はじめに

前回は、売上分析を受け付ける Agent A と、分析を担当する Agent B を別プロセスとして起動し、A2A（Agent2Agent）で連携させました。

そのとき Agent B は、受け取った売上データの抽出・集計・説明をローカル LLM にまとめて依頼していました。小さなサンプルでは動作を確認できますが、completed の絞り込みや金額の合計まで LLM の出力に任せる構成は、正確なデータ処理には向きません。

そこで今回は、Agent B の内部に Python Tool を追加します。LLM は依頼を読んで「売上集計 Tool を使うべきか」「どの条件で集計するか」を判断し、レコードの絞り込み・件数・金額の計算は Python が決定的に実行します。

この記事で扱うのは、前回作成した Agent B の拡張です。Agent A、A2A Server、Agent Card、A2A Client の構成は変更しません。

Tool を使う理由

LLM は文章の理解、依頼の分類、結果の説明には強みがあります。一方、CSV の全行を漏れなく読み、指定条件でフィルターし、整数を正確に合計する仕事は、Python や SQL のほうが適しています。

担当	今回の役割
ローカル LLM	売上集計が必要かを判断し、Tool の引数を選び、結果を説明する
Python Tool	CSV を検証し、`completed` のレコードを抽出し、日付・商品別に件数と金額を集計する
Agent B	LLM と Tool を組み合わせ、A2A の artifact として結果を返す
Agent A	ユーザーとの窓口となり、Agent B へ A2A task を委譲する

ここでいう Tool は、LLM が任意の Python コードを生成して実行する仕組みではありません。開発者があらかじめ実装・登録した aggregate_sales 関数だけを、定義した引数で呼び出せる仕組みです。この制限が、処理の安全性とテスト可能性につながります。

OpenAI Agents SDK では @function_tool を付けた Python 関数を Agent の tools に登録できます。Agent の基本要素はモデル・Tool・Instructions であり、Tool は外部関数や API に Agent の能力を拡張するためのものです。詳しくは OpenAI の Agents ガイドを参照してください。

今回の完成イメージ

ユーザーが第2回と同じ売上データを Agent A に渡すと、処理は次の順で進みます。

1. Agent A が売上分析の依頼を A2A task として Agent B へ送る
2. Agent B の LLM が、日付・商品別の集計には aggregate_sales が必要だと判断する
3. Agents SDK が登録済みの aggregate_sales 関数を呼び出す
4. Python が元の CSV を検証・抽出・集計し、Markdown 表を返す
5. Agent B が Tool の結果を説明付きで最終出力にする
6. Agent B が A2A artifact を返し、Agent A がユーザーへ表示する

今回確認すること

Agent B に function_tool で Python 関数を登録できる
LLM が必要に応じて Tool を呼び出せる
元の入力データは Python で検証し、決定的に集計できる
Tool の結果を Agent B の最終出力と A2A artifact に渡せる
Agent A と A2A の実装を変更せず、Agent B の能力だけを拡張できる

前提環境

前回の記事の agent-a / agent-b プロジェクトを使います。agent-b の依存関係にはすでに openai-agents、openai、a2a-sdk が含まれているため、追加パッケージは不要です。

ローカル LLM サーバーには、前回の Chat Completions API に加えて Tool Calling（function calling） が必要です。モデルだけでなく、OpenAI 互換サーバーが次の往復を実装していることを確認してください。

LLM が tool call を返す
  ↓
クライアントが Tool を実行する
  ↓
tool result を会話へ追加する
  ↓
LLM が最終回答を返す

互換 API の Tool Calling 実装には差があります。まずはモデルとサーバーの組み合わせで、小さな Tool を1つだけ登録して動作確認することを勧めます。本記事では引き続き OpenAIChatCompletionsModel を使います。

Agent B に集計 Tool を追加する

第2回の agent-b を次のように拡張します。

agent-b/
├─ analysis_agent.py  # Tool を持つ Agent B を定義する
├─ sales_tool.py      # CSV を検証・集計する Python Tool を追加する
├─ executor.py        # A2A task を Agent B へ渡す（変更なし）
└─ server.py

Tool が参照する実行時コンテキストを定義する

Tool に売上データ全体を引数として渡すと、LLM は長い CSV を Tool Calling の引数に再出力しなければなりません。転記漏れや入力の改変を避けるため、Agent B が受け取った元の依頼は実行時コンテキストに保持します。

LLM が Tool に渡すのは、target_status と group_by だけです。Python Tool はコンテキストにある元データを直接読みます。

`sales_tool.py` を作成する

次のコードは、CSV の列を検証してから completed の決済だけを日付・商品別に集計します。Tool が返す表の数値は、LLM ではなく Python が作ります。

from __future__ import annotations

import csv
from collections import defaultdict
from dataclasses import dataclass
from io import StringIO
from typing import Literal

from agents import RunContextWrapper, function_tool


@dataclass
class SalesRequestContext:
    """The original A2A request. It is not generated again by the LLM."""

    request_text: str


EXPECTED_FIELDS = ("取引ID", "日付", "商品", "金額（円）", "status")


def read_sales_records(request_text: str) -> list[dict[str, str]]:
    """Extract and validate the CSV section in the user request."""
    lines = request_text.splitlines()
    header = ",".join(EXPECTED_FIELDS)
    try:
        start = lines.index(header)
    except ValueError as exc:
        raise ValueError("売上 CSV のヘッダーが見つかりません") from exc

    reader = csv.DictReader(StringIO("\n".join(lines[start:])))
    if tuple(reader.fieldnames or ()) != EXPECTED_FIELDS:
        raise ValueError("売上 CSV の列が想定と異なります")

    records: list[dict[str, str]] = []
    for line_number, row in enumerate(reader, start=2):
        if set(row) != set(EXPECTED_FIELDS) or any(value is None for value in row.values()):
            raise ValueError(f"CSV の {line_number} 行目の列数が正しくありません")
        try:
            int(row["金額（円）"])
        except ValueError as exc:
            raise ValueError(f"CSV の {line_number} 行目の金額が整数ではありません") from exc
        records.append(row)
    return records


@function_tool
def aggregate_sales(
    context: RunContextWrapper[SalesRequestContext],
    target_status: Literal["completed"],
    group_by: list[Literal["date", "product"]],
) -> str:
    """Aggregate sales records by date and product.

    Use this tool when the request needs record filtering, counts, or sales totals.
    target_status must be "completed" and group_by must be ["date", "product"].
    """
    if target_status != "completed":
        raise ValueError("このサンプルで集計できる status は completed だけです")
    if group_by != ["date", "product"]:
        raise ValueError("このサンプルは date, product の順でのみ集計できます")

    totals: dict[tuple[str, str], list[int]] = defaultdict(lambda: [0, 0])
    for row in read_sales_records(context.context.request_text):
        if row["status"] != target_status:
            continue
        key = (row["日付"], row["商品"])
        totals[key][0] += 1
        totals[key][1] += int(row["金額（円）"])

    table = [
        "| 日付 | 商品 | 決済件数 | 売上金額（円） |",
        "| --- | --- | ---: | ---: |",
    ]
    for (date, product), (count, amount) in sorted(totals.items()):
        table.append(f"| {date} | {product} | {count} | {amount:,} |")

    print(f"[Agent B Tool] Aggregated {len(totals)} date/product groups")
    return "\n".join(table)

RunContextWrapper を引数に取ることで、Tool は Runner.run() ごとに渡した SalesRequestContext を参照できます。この引数は LLM が作る Tool 引数には含まれません。

また、Literal 型により Tool の引数を限定しています。Instructions は重要ですが、最終的な入力検証は Tool 側でも行います。LLM の出力だけを信頼して処理を実行しないことが重要です。

`analysis_agent.py` を変更する

第2回の Agent B に Tool と実行時コンテキストを追加します。Runner.run() の context に元の A2A request を渡す点がポイントです。

"""The OpenAI Agents SDK implementation used internally by Agent B."""

from __future__ import annotations

import os

from agents import Agent, OpenAIChatCompletionsModel, Runner, set_tracing_disabled
from openai import AsyncOpenAI

from sales_tool import SalesRequestContext, aggregate_sales


set_tracing_disabled(True)


def create_analysis_agent() -> Agent[SalesRequestContext]:
    client = AsyncOpenAI(
        base_url=os.environ["LOCAL_LLM_BASE_URL"],
        api_key=os.environ.get("LOCAL_LLM_API_KEY", "not-needed"),
    )
    model = OpenAIChatCompletionsModel(
        model=os.environ["LOCAL_LLM_MODEL"],
        openai_client=client,
    )
    return Agent(
        name="Sales Analysis Specialist",
        instructions=(
            "You are a sales-data analysis specialist. "
            "Use only the records and instructions in the request. "
            "If the request asks to filter records, count transactions, or total sales, "
            "you must call aggregate_sales. "
            "For the sample sales request, call it with target_status='completed' "
            "and group_by=['date', 'product']. "
            "Treat the tool result as authoritative: do not calculate, alter, or invent values. "
            "Return concise Japanese. Include the tool's Markdown table unchanged and "
            "briefly state that cancelled records are excluded."
        ),
        model=model,
        tools=[aggregate_sales],
    )


async def analyze(request_text: str) -> str:
    result = await Runner.run(
        create_analysis_agent(),
        input=request_text,
        context=SalesRequestContext(request_text=request_text),
    )
    if not result.final_output:
        raise RuntimeError("The analysis agent returned no final output.")
    return result.final_output

executor.py は第2回のままでかまいません。A2A のメッセージ本文を analyze(request_text) に渡す既存の経路の中で、Agent B が Tool を使うようになったためです。

実行する

前回と同じように、ローカル LLM サーバー、Agent B、Agent A の順に起動します。

# Terminal 1: Agent B
cd agent-b
.venv\Scripts\Activate.ps1
$env:LOCAL_LLM_BASE_URL = "http://127.0.0.1:11434/v1"
$env:LOCAL_LLM_MODEL = "<Tool Calling に対応するモデル ID>"
python server.py

# Terminal 2: Agent A
cd agent-a
.venv\Scripts\Activate.ps1
$env:AGENT_B_URL = "http://127.0.0.1:9999"
$env:AGENT_B_TIMEOUT_SECONDS = "300"
$env:LOCAL_LLM_BASE_URL = "http://127.0.0.1:11434/v1"
$env:LOCAL_LLM_MODEL = "<Tool Calling に対応するモデル ID>"
Get-Content -Raw -Encoding utf8 request.txt | python main.py

request.txt は第2回で使ったものをそのまま使えます。completed だけを対象に、日付・商品別の決済件数と売上金額を求める依頼です。

実行結果

Agent A は Agent B の Agent Card を取得し、A2A task を送信して artifact を受け取りました。次は、その1回の実行結果です。

[Agent A] Retrieved Agent B's Agent Card

Name        : Sales Analysis Agent
Description : Uses a deterministic Python tool to aggregate supplied sales data.
Version     : 0.2.0
Interface   : http://127.0.0.1:9999 (JSONRPC 1.0)
Skill       : sales_analysis

[Agent A] Sending an A2A task to Agent B
[Agent A] Received Agent B's analysis artifact
[Agent A] Final answer

| 日付 | 商品 | 決済件数 | 売上金額（円） |
| --- | --- | ---: | ---: |
| 2026-08-01 | ノート | 2 | 600 |
| 2026-08-01 | ボールペン | 1 | 150 |
| 2026-08-01 | マグカップ | 1 | 1,800 |
| 2026-08-02 | ノート | 1 | 300 |
| 2026-08-02 | ボールペン | 2 | 300 |
| 2026-08-02 | マグカップ | 1 | 1,800 |
| 2026-08-03 | ノート | 2 | 600 |
| 2026-08-03 | ボールペン | 1 | 150 |
| 2026-08-04 | ノート | 1 | 300 |
| 2026-08-04 | ボールペン | 3 | 450 |
| 2026-08-05 | ノート | 3 | 900 |
| 2026-08-05 | マグカップ | 1 | 1,800 |
| 2026-08-06 | ボールペン | 2 | 300 |
| 2026-08-06 | マグカップ | 1 | 1,800 |
| 2026-08-07 | ノート | 2 | 600 |
| 2026-08-07 | ボールペン | 1 | 150 |
| 2026-08-07 | マグカップ | 1 | 1,800 |
| 2026-08-08 | ボールペン | 4 | 600 |
| 2026-08-08 | マグカップ | 1 | 1,800 |

2026-08-03 のマグカップと 2026-08-08 のノートは cancelled のため、最終表には含まれていません。

ここで重要なのは表の見た目ではなく、数値の生成経路です。今回の Agent B は集計が必要だと判断すると aggregate_sales を Tool として呼び出します。Tool は元の CSV を検証し、completed のレコードを Python で絞り込み・加算してこの表を返します。したがって、表中の件数と売上金額は LLM がトークン列として計算した値ではありません。

LLM と Tool の境界を確認する

この例で正確性を担保しているのは、Instructions ではありません。Instructions は LLM が Tool を選ぶ確率を高めますが、集計の正しさそのものを保証しません。

正確性に関わる部分は Tool に閉じ込めます。

CSV のヘッダーと列数を検証する
金額を int として読み取る
status == "completed" を Python の条件式で判定する
件数と金額を Python で加算する
表を Python が生成する

一方、最終的な自然言語の説明は LLM が行います。今回の実行では Agent A の最終出力にも Tool が作成した表と同じ数値が返りました。ただし、現在の Agent A は Agent B の artifact を受け取った後に LLM で最終回答を生成するため、数値をプログラム上で不変にする保証まではありません。

数値を画面や後続システムへ厳密に渡す必要がある場合は、Tool が JSON などの構造化データを返し、呼び出し元がその値を直接使う設計を検討してください。本記事では読みやすさを優先し、Tool が Markdown 表を返し、Agent B にはその表を変更しないよう指示しています。

実装時の注意点

Tool Calling に失敗する場合

原因を Agent の Instructions だけに求めず、まず次を確認します。

ローカル LLM とサーバーが Tool Calling をサポートしているか
モデル名に Tool Calling 用のテンプレートや設定が必要ではないか
サーバーが Tool の結果を含む Chat Completions の複数ターンを処理できるか
Tool の JSON Schema がモデルにとって複雑すぎないか

最初は Tool を1つにし、引数も少なく保つと原因を切り分けやすくなります。

Tool は小さく、入力を検証する

aggregate_sales は「売上 CSV を日付・商品別に集計する」ことだけを担当します。ファイル読み込み、任意 SQL の実行、外部 API 呼び出しなどを1つの万能 Tool にまとめないほうが、権限・テスト・失敗時の扱いを明確にできます。

書き込み、送信、削除、決済のような副作用を持つ Tool では、引数の検証に加えて認可や人による承認を設計してください。今回の Tool は入力を読み取って集計するだけで、外部状態を変更しません。

集計 Tool を単体テストする

Agent を経由せずに read_sales_records() と集計ロジックをテストできるようにしておくと、モデルの振る舞いと Python の不具合を分けて調査できます。たとえば、cancelled を含む入力で件数・金額・出力行数を固定値で検証します。

まとめ

第2回の Agent B に Python Tool を追加した
LLM は Tool が必要かとその引数を判断し、CSV の集計は Python が実行する
Tool には元の依頼を実行時コンテキストとして渡し、LLM に CSV を再出力させない
入力検証と集計を決定的なコードに置くことで、LLM だけに数値の正確性を委ねない
Agent A と A2A の構成を変えずに、Agent B の分析能力を拡張できた

参考