MCP・連携参考
Tool Use 設計と MCP サーバー実装の参考。RAG・実装参考 と併せて読む。
Tool Use・MCP 実装
LLMにツール実行能力を与えるTool Use設計と、MCPサーバー実装の実例。DB・API・ファイル操作をエージェントから安全に呼ぶ。
Tool Use 設計原則
① 説明を詳細に書く
ツールの description が不明確だと LLM が誤用する。いつ使うか・何を返すか・制限事項を明記する。
Bad vs Good
// ❌ 不明確
"description": "DBを検索する"
// ✅ 明確
"description": "注文DBから顧客IDで
注文履歴を検索する。最大30件返す。
存在しない場合は空配列を返す。"② スキーマを厳密に
input_schema で型・enum・必須項目を明示する。LLMが自由に値を作れる余地を減らすことでハルシネーションを防ぐ。
enum の活用
"status": {
"type": "string",
"enum": ["PENDING","CONFIRMED",
"SHIPPED","CANCELLED"],
"description": "注文ステータス"
}③ エラーを構造化
ツールがエラーを返す際はLLMが次のアクションを決められる形式で返す。生の例外スタックトレースは渡さない。
エラー形式
// ✅ 構造化エラー
{"error": "NOT_FOUND",
"message": "注文ID ord-123 は存在しません",
"suggestion": "get_orders で一覧を確認してください"}実装例 — 実務でよく使うツールセット
PostgreSQL 検索ツール(安全なSQL生成付き)
import anthropic
import asyncpg
import json
from typing import Any
# ── ツール定義 ────────────────────────────────────────────────
DB_TOOLS = [
{
"name": "search_orders",
"description": "注文データベースを検索する。顧客ID・ステータス・日付範囲での絞り込みが可能。最大50件を返す。",
"input_schema": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "顧客ID(例: cust_123)。指定しない場合は全顧客対象"
},
"status": {
"type": "string",
"enum": ["PENDING", "CONFIRMED", "SHIPPED", "CANCELLED"],
"description": "注文ステータスで絞り込み"
},
"date_from": {
"type": "string",
"description": "検索開始日 (YYYY-MM-DD形式)"
},
"date_to": {
"type": "string",
"description": "検索終了日 (YYYY-MM-DD形式)"
},
"limit": {
"type": "integer",
"default": 20,
"maximum": 50,
"description": "取得件数(最大50)"
}
}
}
},
{
"name": "get_order_detail",
"description": "注文IDで注文の詳細情報(明細・配送状況・支払い情報)を取得する",
"input_schema": {
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "注文ID(例: ord_abc123)"}
},
"required": ["order_id"]
}
}
]
# ── ツール実行(パラメータバインディングでSQLインジェクション防止) ─
async def execute_search_orders(params: dict, db: asyncpg.Connection) -> str:
conditions = ["1=1"]
args = []
if customer_id := params.get("customer_id"):
args.append(customer_id)
conditions.append(f"customer_id = ${len(args)}")
if status := params.get("status"):
args.append(status)
conditions.append(f"status = ${len(args)}")
if date_from := params.get("date_from"):
args.append(date_from)
conditions.append(f"created_at >= ${len(args)}::date")
if date_to := params.get("date_to"):
args.append(date_to)
conditions.append(f"created_at <= ${len(args)}::date")
limit = min(params.get("limit", 20), 50) # 上限を強制
sql = f"""
SELECT id, customer_id, status, total_amount, created_at
FROM orders
WHERE {' AND '.join(conditions)}
ORDER BY created_at DESC
LIMIT {limit}
"""
rows = await db.fetch(sql, *args)
if not rows:
return json.dumps({"orders": [], "message": "条件に一致する注文が見つかりませんでした"})
return json.dumps({
"orders": [dict(row) for row in rows],
"count": len(rows),
}, default=str) # datetime を文字列化
# ── エージェント実行ループ ─────────────────────────────────────
async def run_db_agent(question: str, db: asyncpg.Connection) -> str:
client = anthropic.Anthropic()
messages = [{"role": "user", "content": question}]
while True:
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
tools=DB_TOOLS,
system="あなたは注文管理DBのアシスタントです。ユーザーの質問に答えるためにツールを使ってDBを検索してください。",
messages=messages,
)
if response.stop_reason == "end_turn":
return next(b.text for b in response.content if hasattr(b, "text"))
tool_results = []
for block in response.content:
if block.type == "tool_use":
if block.name == "search_orders":
result = await execute_search_orders(block.input, db)
elif block.name == "get_order_detail":
result = json.dumps({"error": "NOT_IMPLEMENTED"})
else:
result = json.dumps({"error": "UNKNOWN_TOOL"})
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
💡 DBツールの安全対策:① SQLはパラメータバインディング($1, $2)で SQLインジェクション完全防止 ② limit を強制上限でキャップ ③ 書き込み系ツール(UPDATE/DELETE)は確認ステップを必ず挟む ④ 返り値の JSON は default=str で型変換エラーを防ぐ
外部API連携ツール(Slack・Calendar・GitHub)
import httpx
import json
# ── Slack投稿ツール ───────────────────────────────────────────
SLACK_TOOL = {
"name": "post_slack_message",
"description": "指定Slackチャンネルにメッセージを投稿する。承認が必要な重要メッセージに使う。",
"input_schema": {
"type": "object",
"properties": {
"channel": {"type": "string", "description": "チャンネル名(例: #general)"},
"message": {"type": "string", "description": "投稿するメッセージ内容"},
"mention_users": {
"type": "array",
"items": {"type": "string"},
"description": "メンションするユーザーIDのリスト"
}
},
"required": ["channel", "message"]
}
}
async def post_slack_message(channel: str, message: str, mention_users: list = None) -> str:
text = message
if mention_users:
mentions = " ".join(f"<@{uid}>" for uid in mention_users)
text = f"{mentions}
{message}"
async with httpx.AsyncClient() as client:
response = await client.post(
"https://slack.com/api/chat.postMessage",
headers={"Authorization": f"Bearer {SLACK_TOKEN}"},
json={"channel": channel, "text": text},
)
data = response.json()
if not data.get("ok"):
return json.dumps({"error": data.get("error"), "suggestion": "チャンネル名とbot権限を確認してください"})
return json.dumps({"success": True, "ts": data["ts"], "channel": channel})
# ── GitHub Issue作成ツール ────────────────────────────────────
GITHUB_TOOL = {
"name": "create_github_issue",
"description": "GitHubリポジトリにIssueを作成する。バグ報告・機能要望・タスク管理に使う。",
"input_schema": {
"type": "object",
"properties": {
"repo": {"type": "string", "description": "リポジトリ名(例: org/repo)"},
"title": {"type": "string", "description": "Issueのタイトル"},
"body": {"type": "string", "description": "Issue本文(Markdown対応)"},
"labels": {"type": "array", "items": {"type": "string"}, "description": "ラベルのリスト"},
"assignees": {"type": "array", "items": {"type": "string"}, "description": "担当者GitHubユーザー名"}
},
"required": ["repo", "title", "body"]
}
}
async def create_github_issue(repo: str, title: str, body: str,
labels: list = None, assignees: list = None) -> str:
async with httpx.AsyncClient() as client:
response = await client.post(
f"https://api.github.com/repos/{repo}/issues",
headers={
"Authorization": f"Bearer {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3+json",
},
json={"title": title, "body": body,
"labels": labels or [], "assignees": assignees or []},
)
if response.status_code == 201:
data = response.json()
return json.dumps({"success": True, "issue_url": data["html_url"], "issue_number": data["number"]})
return json.dumps({"error": "CREATION_FAILED", "status": response.status_code})
💡 外部API連携の注意点:① 書き込み系API(Slack投稿・Issue作成)は人間確認ステップを挟む設計にする ② レート制限に引っかかった場合のエラーを suggestion 付きで返す ③ 機密情報(APIキー)はツールの input/output に含めない
MCPサーバー実装(TypeScript)
// MCPサーバー: エージェントからDBやAPIをツールとして呼べるようにする
// npm install @modelcontextprotocol/sdk
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const server = new McpServer({
name: "order-management-mcp",
version: "1.0.0",
});
// ── ツール登録 ────────────────────────────────────────────────
server.tool(
"search_orders",
"注文データベースを検索する",
{
customer_id: z.string().optional().describe("顧客ID"),
status: z.enum(["PENDING", "CONFIRMED", "SHIPPED", "CANCELLED"]).optional(),
limit: z.number().int().min(1).max(50).default(20),
},
async ({ customer_id, status, limit }) => {
try {
const conditions: string[] = [];
const params: unknown[] = [];
if (customer_id) { params.push(customer_id); conditions.push(`customer_id = $${params.length}`); }
if (status) { params.push(status); conditions.push(`status = $${params.length}`); }
const where = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
const rows = await db.query(
`SELECT id, customer_id, status, total_amount, created_at FROM orders ${where} ORDER BY created_at DESC LIMIT $${params.length + 1}`,
[...params, limit]
);
return {
content: [{
type: "text",
text: JSON.stringify({ orders: rows.rows, count: rows.rowCount }),
}],
};
} catch (err) {
return {
content: [{ type: "text", text: JSON.stringify({ error: "DB_ERROR", message: String(err) }) }],
isError: true,
};
}
}
);
// ── リソース登録(読み取り専用データ) ────────────────────────
server.resource(
"order-schema",
"postgres://orders/schema",
async (uri) => ({
contents: [{
uri: uri.href,
mimeType: "application/json",
text: JSON.stringify({
table: "orders",
columns: ["id", "customer_id", "status", "total_amount", "created_at"],
statuses: ["PENDING", "CONFIRMED", "SHIPPED", "CANCELLED"],
}),
}],
})
);
// ── サーバー起動(stdio transport)───────────────────────────
const transport = new StdioServerTransport();
await server.connect(transport);
// ── claude_desktop_config.json に追加する設定 ─────────────────
// {
// "mcpServers": {
// "order-management": {
// "command": "node",
// "args": ["/path/to/mcp-server/dist/index.js"],
// "env": { "DATABASE_URL": "postgresql://..." }
// }
// }
// }
💡 MCPの使い分け:① Claude Desktop/Cowork に接続してローカル開発環境で使う ② プロダクションエージェントには Anthropic Tool Use を直接実装する方がシンプル ③ MCPは「再利用可能なツールサーバー」として複数プロジェクトから共有できる利点がある
マルチエージェント
単一の ReAct ループでは長タスク・多ドメイン統合が難しいとき、役割分担した複数エージェントが協調する。flow-18 マルチエージェント協調 は Cursor / Claude Code の開発セッション運用が正本。本節は プロダクト / バックエンド実装(LangGraph・SDK)向け。Agentic RAG との統合は RAG 参考 — Agentic RAG。
ReAct ループ
Thought → Action(ツール)→ Observation を繰り返す基本パターン。1エージェント・短〜中タスク向き。LangChain / Anthropic SDK の標準形。
ループ
Thought: 売上データが必要
Action: get_sales(month)
Obs: total 1200000
Thought: グラフ化して報告
Action: generate_chart(...)Plan-and-Execute
先に全体計画を立て、サブタスクを順次実行。複雑な多ステップに向く。LangGraph StateGraph や extended thinking と相性が良い。
使い所
「競合調査レポートを作って」
→ Plan: 検索×5, 要約, 比較表, 執筆
→ 各ステップを順次実行Orchestrator + Subagents
Orchestrator が Subagent に指示し、並列実行・結果統合を行う。Researcher / Coder / Reviewer など役割固定が典型。
構成
Orchestrator
├── Researcher(検索)
├── Coder(実装)
└── Reviewer(レビュー)| パターン | 向くユースケース | 主な実装 | 難易度 |
|---|---|---|---|
| シンプル RAG | 社内文書 Q&A・FAQ | LlamaIndex / LangChain | ★★☆☆☆ |
| Tool-use Agent | DB 検索・API 連携・レポート生成 | Anthropic SDK / AI SDK | ★★★☆☆ |
| Plan-and-Execute | 複雑な調査・コード生成・長タスク | LangGraph / Claude SDK | ★★★★☆ |
| マルチエージェント | 並列リサーチ・専門役割分担・大規模自動化 | LangGraph Supervisor / asyncio Orchestrator | ★★★★★ |
flowchart LR
subgraph pipe [Pipeline]
P1[Research] --> P2[Analyze] --> P3[Write] --> P4[Review]
end
subgraph fan [Fanout]
F0[Task] --> F1[Item1]
F0 --> F2[Item2]
F0 --> F3[Item3]
F1 --> FM[Merge]
F2 --> FM
F3 --> FM
end
| パターン | 向く場面 | 注意点 |
|---|---|---|
| パイプライン | 段階依存が明確(調査→執筆→レビュー) | 受け渡しスキーマを固定 |
| ファンアウト | 独立アイテムの並列処理 | 集約ロジックを先に定義 |
| スペシャリスト | 多ドメイン統合(市場・技術・財務) | 出力形式を標準化 |
Supervisor パターン(LangGraph)
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END
class TeamState(TypedDict):
task: str
next_worker: str
research: str
draft: str
final: str
def supervisor(state: TeamState) -> TeamState:
# LLM が次の worker を選ぶ(structured output 推奨)
worker = llm_pick_worker(state["task"], state)
return {**state, "next_worker": worker}
def researcher(state: TeamState) -> TeamState:
out = run_research_agent(state["task"]) # RAG tool 可
return {**state, "research": out}
def writer(state: TeamState) -> TeamState:
out = run_writer_agent(state["research"])
return {**state, "draft": out}
def route(state: TeamState) -> Literal["researcher", "writer", "end"]:
w = state["next_worker"]
if w == "researcher":
return "researcher"
if w == "writer":
return "writer"
return "end"
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route, {
"researcher": "researcher", "writer": "writer", "end": END,
})
graph.add_edge("researcher", "supervisor")
graph.add_edge("writer", "supervisor")
team = graph.compile()
💡 Researcher に Agentic RAG の search ツールを渡すと、調査専門エージェントが自律的に retrieve / grade できる。
Orchestrator + Subagents(asyncio)
import anthropic
import asyncio
client = anthropic.Anthropic()
async def researcher_agent(topic: str) -> str:
resp = client.messages.create(
model="claude-haiku-4-5",
max_tokens=2048,
system="調査専門。重要ポイントを箇条書きで返す。",
messages=[{"role": "user", "content": f"調査: {topic}"}],
)
return resp.content[0].text
async def writer_agent(research: str) -> str:
resp = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
system="技術ブログ調で記事化する。",
messages=[{"role": "user", "content": research}],
)
return resp.content[0].text
async def orchestrator(topic: str) -> str:
# ファンアウト: 並列調査
parts = await asyncio.gather(
researcher_agent(f"{topic} - 技術"),
researcher_agent(f"{topic} - ビジネス"),
)
combined = "
".join(parts)
return await writer_agent(combined)
# asyncio.run(orchestrator("AIエージェント開発"))
💡 サブタスクは軽量モデル(Haiku)、統合・執筆は高性能モデル(Sonnet)。max_iterations とタイムアウトで暴走を防ぐ。
Cursor / Claude Code — 開発セッションでの委譲
# メイン会話は実装に集中。広い調査は Task に委譲する。
# プロンプト例(Cursor Agent)
「src/domain/ のみ調査。呼び出し元一覧を { paths, summary } JSON で返す。
全文ダンプは不要。」
# subagent_type の目安
# explore — コードベース探索(readonly)
# generalPurpose — 汎用・複数ステップ
# shell — git / コマンド実行
# 受け渡し契約(flow-18 と同じ)
# { "paths": ["src/foo.ts"], "summary": "3行以内" }
プロダクト向けチェックリスト
- ゴールと完了条件を文章化(テスト・評価指標を含む)
- 役割分割 — 汎用エージェント1体より Researcher / Writer / Reviewer など2〜3体から
- 出力スキーマ固定 — JSON Schema / Pydantic で次段がパースできる形に
- 2体パイプラインで動作確認してからファンアウトを増やす
- 観測 — Langfuse / LangSmith でトレース、トークン・latency をダッシュボード化
よくある失敗と対策
| 失敗パターン | 対策 |
|---|---|
| ❌ 汎用エージェントだらけ — 役割が曖昧で特化のメリットが消える | ✅ system prompt とツールセットを役割ごとに分離。Orchestrator のみ全体を見る |
| ❌ 出力形式未標準化 — 次段が free text をパースできず連鎖失敗 | ✅ structured output / JSON mode。受け渡し契約を README または型定義に固定 |
| ❌ 並列を早まりで増やす — 観測不能・重複作業・コスト肥大 | ✅ まず直列パイプライン2体。ファンアウトは集約ロジック確定後 |
マルチエージェントを AI に依頼するプロンプト例
競合調査レポート
競合3社の調査レポートを自動生成するマルチエージェントを設計してください。 ・Orchestrator が全体管理 ・Researcher を asyncio.gather で並列実行 ・Writer が最終レポート、Reviewer が品質チェック ・各 Subagent の入出力を JSON スキーマで定義
RAG 統合
LangGraph Supervisor に Researcher ノードを追加してください。 Researcher は Agentic RAG(retrieve → grade → rewrite)をサブグラフとして持つ。 Writer は research フィールドのみから執筆する。
AIへの指示パターン
Tool Use実装をAIに依頼するプロンプト例
ツール定義生成
以下のAPIドキュメントから Anthropic Tool Use 形式のツール定義を生成してください。
【API】Stripe の charge・refund・list_charges エンドポイント
条件:① description は「いつ使うか」を明記 ② input_schema に enum・format・description を付ける ③ 返り値の型もコメントで示す
MCPサーバー生成
以下のDBスキーマに対するMCPサーバーを TypeScript で実装してください。
【DB】PostgreSQL(users, orders, products テーブル)
・search/get_by_id/create の3ツールを実装
・Zod でバリデーション ・エラーは構造化して返す ・stdio transport