はじめに
前回は、売上分析を受け付ける 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 の分析能力を拡張できた