AI Agent ツールと外部連携
この章では、Agent がどのように外部世界と対話するかを紹介します。
ツール呼び出し機能は、Agent が外部環境と接続するための鍵です。
さまざまなツールを統合することで、Agent はコードの実行、API へのアクセス、ファイルシステムの操作などができます。
Computer Use
Computer Use により、Agent は人間のようにコンピューター画面を操作できます。
ブラウザ、デスクトップアプリなどのグラフィカルインターフェースを含みます。
これは汎用人工知能（AGI）を実現するための重要な一歩です。
コア機能
スクリーンショット解析：画面上のコンテンツを理解し、操作可能な要素を識別します。
GUI 要素認識：ボタン、入力欄、メニューなどのインターフェース要素を識別します。
マウス/キーボード制御：人間の操作をシミュレートし、クリックや入力などの動作を実行します。
ブラウザ自動化：Web ページのナビゲーション、フォーム入力、検索など。
動作原理
Computer Use のワークフローは、「観察、理解、決定、実行」とまとめることができます。
最初に、スクリーンショット。現在の画面またはウィンドウの内容を取得します。
次に、視覚分析。視覚モデルを使用してスクリーンショットを分析し、インターフェース要素と状態を識別します。
その次に、決定。タスクの目標に基づいて、次の操作を決定します。
最後に、実行。マウスクリックやキーボード入力などの動作を実行します。
タスクが完了するまで繰り返し実行します。
コード実装
Computer Use Agent の基本実装
"""
Computer Use Agent の実装
人間のようにコンピューター画面を操作できる
"""
def __init__(self, vision_model, action_executor, planner):
# 視覚モデル：スクリーンショットを分析
self.vision_model = vision_model
# 動作実行器：マウス・キーボード操作を実行
self.action_executor = action_executor
# プランナー：次の行動を決定
self.planner = planner
# 最大実行ステップ数
self.max_steps = 100
def observe(self, screenshot):
"""
スクリーンショットを解析し、操作可能な要素を識別する
:param screenshot: スクリーンショット画像
:return: インターフェース要素のリスト
"""
# 視覚モデルを使用してスクリーンショットを分析
analysis = self.vision_model.analyze(screenshot)
# 識別したインターフェース要素を返す
# 各要素には、種類、位置、内容、操作可能性が含まれます
return analysis.elements
def act(self, action):
"""
アクションを実行
:param action: アクションの説明。例: {"type": "click", "x": 100, "y": 200}
"""
return self.action_executor.execute(action)
def run(self, task):
"""
タスクループを実行
:param task: タスクの説明
:return: タスクの結果
"""
# タスク状態を初期化
self.planner.set_task(task)
for step in range(self.max_steps):
# ステップ1：現在の画面をキャプチャ
screenshot = self.get_screen()
# ステップ2：観察 - インターフェース要素を識別
elements = self.observe(screenshot)
# ステップ3：決定 - 次のアクションを決定
action = self.planner.decide_action(elements)
# タスクが完了したかどうかを確認
if action.is_final:
return action.result
# ステップ4：アクションを実行
self.act(action)
# オプション：インターフェースの更新を待機
self.wait_for_update()
return "最大ステップ数に達しました"
class VisionModel:
"""視覚モデル：スクリーンショットを分析"""
def analyze(self, screenshot):
"""
スクリーンショットを分析
インターフェース要素のリストを返す
"""
# マルチモーダルモデルを使用して分析
prompt = """
このスクリーンショットを分析し、操作可能なすべてのインターフェース要素を識別してください。
ボタン、入力欄、リンク、メニューなどを含みます。
各要素について、以下を提供してください：
1. 種類（button、input、link など）
2. 位置（境界ボックス座標）
3. 内容（ボタンのテキスト、入力欄のプレースホルダーなど）
4. 操作可能性（表示されるかどうか、有効かどうか）
"""
result = self.vision_model.analyze_image(screenshot, prompt)
return ScreenAnalysisResult(elements=result.elements)
class ActionExecutor:
"""動作実行器：マウス・キーボード操作を実行"""
def execute(self, action):
"""
アクションを実行
:param action: アクションオブジェクト
"""
if action.type == "click":
self.mouse.click(action.x, action.y)
elif action.type == "type":
self.keyboard.type_text(action.text)
elif action.type == "scroll":
self.mouse.scroll(action.direction, action.amount)
elif action.type == "press":
self.keyboard.press_key(action.key)
return ActionResult(success=True)
class Planner:
"""プランナー：次の行動を決定"""
def __init__(self, llm):
self.llm = llm
self.task = None
self.history = []
def set_task(self, task):
"""現在のタスクを設定"""
self.task = task
self.history = []
def decide_action(self, elements):
"""
現在のインターフェース状態に基づいて次の行動を決定する
"""
prompt = f"""
現在のタスク：{self.task}
実行済みの履歴アクション：
{self.history}
現在のインターフェース上の操作可能な要素：
{elements}
次のアクションを決定してください。
タスクが完了している場合は、is_final=True を返してください。
それ以外の場合は、実行するアクション（type、座標、パラメーターなど）を返してください。
"""
response = self.llm.generate(prompt)
return Action.parse(response)
応用シナリオ
Computer Use は特に以下のシナリオに適しています：
Web 自動化：フォームの自動入力、データの収集、Web 操作の実行など。
デスクトップアプリ操作：ファイルを開く、ドキュメントを編集、ソフトウェアを操作など。
テスト自動化：UI テスト、回帰テストの自動実行など。
注意：Computer Use は現在も発展途上であり、実行速度が遅い、たまにエラーが発生するなどの問題があります。明確な API があるシナリオでは、Computer Use よりも API を直接呼び出す方が一般的に効率的です。
MCP プロトコル詳解
MCP（Model Context Protocol）はオープン標準プロトコルです。
これにより、AI モデルは外部ツールやデータソースと安全に接続できます。
MCP の設計目標は、AI 分野の「USB インターフェース」になることです。
コア設計思想
標準化：統一されたプロトコル形式により、異なる Agent とツールが相互運用できます。
セキュリティ：明確な権限制御により、Agent は許可されたリソースにのみアクセスできます。
拡張性：新しいツールやデータソースを追加しやすいこと。
アーキテクチャコンポーネント
MCP Host：AI アプリケーションを実行するホスト環境。例: Claude Desktop、AI コーディングアシスタントなど。
MCP Client：MCP Server と 1:1 接続を維持するクライアント。
MCP Server：ツールとリソースを提供するサーバープログラム。
通信フロー
ステップ1、接続確立。Client と Server が接続を確立し、能力情報を交換します。
ステップ2、ツール発見。Client は Server がどのツールとリソースを提供するかを照会します。
ステップ3、ツール呼び出し。Client がツール呼び出しリクエストを送信し、Server が実行して結果を返します。
ステップ4、リソースアクセス。Client は Server が提供するリソースを読み書きできます。
MCP Server 実装例
MCP Server の実装
from mcp.types import Tool, Resource
import asyncio
# MCP Server インスタンスを作成
app = Server("filesystem")
# ファイルシステムツールを定義
@app.list_tools()
async def list_tools():
"""利用可能なすべてのツールを一覧表示"""
return [
Tool(
name="read_file",
description="ファイルの内容を読み取る",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ファイルパス"
},
"encoding": {
"type": "string",
"description": "ファイルエンコーディング。デフォルトは utf-8",
"default": "utf-8"
}
},
"required": ["path"]
}
),
Tool(
name="write_file",
description="ファイルの内容を書き込む",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ファイルパス"
},
"content": {
"type": "string",
"description": "ファイルの内容"
}
},
"required": ["path", "content"]
}
),
Tool(
name="list_directory",
description="ディレクトリの内容を一覧表示",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "ディレクトリパス"
}
}
}
)
]
@app.call_tool()
async def call_tool(name, arguments):
"""ツール呼び出しを実行"""
if name == "read_file":
return await read_file(arguments["path"], arguments.get("encoding", "utf-8"))
elif name == "write_file":
return await write_file(arguments["path"], arguments["content"])
elif name == "list_directory":
return await list_directory(arguments["path"])
else:
raise ValueError(f"Unknown tool: {name}")
async def read_file(path, encoding):
"""ファイルを読み取る"""
with open(path, "r", encoding=encoding) as f:
content = f.read()
return [{"type": "text", "text": content}]
async def write_file(path, content):
"""ファイルを書き込む"""
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return [{"type": "text", "text": "File written successfully"}]
async def list_directory(path):
"""ディレクトリを一覧表示"""
import os
entries = os.listdir(path)
return [{"type": "text", "text": "\n".join(entries)}]
# サーバーを実行
if __name__ == "__main__":
import mcp.server.stdio
asyncio.run(mcp.server.stdio.serve(app))
API 統合パターン
Agent が外部 API を統合するには、一般的に3つの方法があります。
REST API は最も一般的に使用される Web API スタイルです。
GraphQL はより柔軟なデータクエリ機能を提供します。
gRPC は高性能シナリオのリモートプロシージャコールに適しています。
REST API 統合
REST（Representational State Transfer）は Web API の設計スタイルです。
HTTP メソッド（GET、POST、PUT、DELETE）を使用して操作します。
REST API クライアント実装
from typing import Dict, Any, Optional
class RESTAPIClient:
"""
REST API クライアント
HTTP リクエストをカプセル化し、シンプルな呼び出しインターフェースを提供
"""
def __init__(self, base_url: str, headers: Optional[Dict] = None):
"""
API クライアントを初期化
:param base_url: API 基本 URL
:param headers: デフォルトのリクエストヘッダー
"""
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
# デフォルトのリクエストヘッダーを設定
default_headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
if headers:
default_headers.update(headers)
self.session.headers.update(default_headers)
def call(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
data: Optional[Dict] = None,
**kwargs
) -> Dict[Any, Any]:
"""
API リクエストを送信
:param method: HTTP メソッド（GET、POST、PUT、DELETE）
:param endpoint: API エンドポイント
:param params: URL クエリパラメータ
:param data: リクエストボディデータ
:return: レスポンスデータ
"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.request(
method=method.upper(),
url=url,
params=params,
json=data,
**kwargs
)
# HTTP ステータスコードを確認
response.raise_for_status()
# レスポンスを解析
if response.content:
return response.json()
return {}
def get(self, endpoint: str, params: Optional[Dict] = None):
"""GET リクエスト"""
return self.call("GET", endpoint, params=params)
def post(self, endpoint: str, data: Dict):
"""POST リクエスト"""
return self.call("POST", endpoint, data=data)
def put(self, endpoint: str, data: Dict):
"""PUT リクエスト"""
return self.call("PUT", endpoint, data=data)
def delete(self, endpoint: str, params: Optional[Dict] = None):
"""DELETE リクエスト"""
return self.call("DELETE", endpoint, params=params)
class APIAgent:
"""
API Agent
自然言語で外部 API を呼び出す
"""
def __init__(self, api_client: RESTAPIClient):
self.api_client = api_client
self.tool_descriptions = self.define_tools()
def define_tools(self):
"""
Agent が利用可能なツールを定義
ツールの説明を返し、LLM が呼び出し方法を理解できるようにする
"""
return [
{
"name": "get_weather",
"description": "指定した都市の天気情報を取得",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "都市名（例：Beijing、Shanghai）"
}
},
"required": ["city"]
}
},
{
"name": "get_forecast",
"description": "指定した都市の天気予報を取得",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "都市名"
},
"days": {
"type": "integer",
"description": "予報日数、デフォルトは 3",
"default": 3
}
},
"required": ["city"]
}
}
]
def execute_tool(self, tool_name: str, arguments: Dict):
"""
ツール呼び出しを実行
:param tool_name: ツール名
:param arguments: ツールパラメータ
:return: 実行結果
"""
if tool_name == "get_weather":
return self.api_client.get(f"/weather/{arguments['city']}")
elif tool_name == "get_forecast":
params = {"days": arguments.get("days", 3)}
return self.api_client.get(f"/forecast/{arguments['city']}", params=params)
else:
raise ValueError(f"Unknown tool: {tool_name}")
API 統合のベストプラクティス
エラーハンドリング：ネットワークエラー、タイムアウト、サービス利用不可などを適切に処理します。
リトライメカニズム：一時的なエラーに対して、指数バックオフリトライを実装します。
レート制限制御：API のレート制限を遵守し、ブロックを回避します。
キャッシュ戦略：頻繁にリクエストされるデータをキャッシュし、API 呼び出しを削減します。
セキュリティ考慮事項：API キーなどの機密情報はハードコードせず、環境変数で管理します。
章のまとめ
この章では、Agent が外部世界と対話するためのコア技術を紹介しました。
Computer UseAgent が GUI インターフェースを操作できるようにし、グラフィカルアプリとの対話が必要なシナリオに適しています。
MCP プロトコル標準化されたツール接入方法を提供し、Agent 時代の「USB インターフェース」です。
API 統合は最も一般的な外部システム統合方法であり、REST、GraphQL、gRPC などを含みます。
適切な統合方法の選択は、具体的なシナリオに応じて行う必要があります。
グラフィカルインターフェース操作には、Computer Use が汎用ソリューションです。
ツールの標準化された接入には、MCP が将来の发展方向です。
Web API 統合には、REST API が依然として最も主流の選択肢です。
その他の拡張