Gemini で自律エージェントを動かす最小構成 — ツールループ設計で学んだこと

from google import genai
from google.genai import types
 
# GEMINI_API_KEY 環境変数を自動で読み込みます
client = genai.Client()
 
def get_article_stats(slug: str) -> dict:
    """指定したスラッグの記事の直近7日間の閲覧数とクリック数を返します。"""
    # 実際には GA4 / Search Console のデータストアを引きます
    return {"slug": slug, "views": 1280, "clicks": 96}
 
def list_pending_drafts() -> list[str]:
    """公開待ちの下書きスラッグ一覧を返します。"""
    return ["gemini-batch-api-notes", "image-asset-pipeline"]
 
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="公開待ちの下書きを確認して、直近の人気記事の傾向と合わせて公開順の提案をしてください。",
    config=types.GenerateContentConfig(
        tools=[get_article_stats, list_pending_drafts],
    ),
)
 
print(response.text)

from google import genai
from google.genai import types
 
client = genai.Client()
 
publish_tool = types.Tool(function_declarations=[
    types.FunctionDeclaration(
        name="publish_article",
        description="指定スラッグの記事を公開します。破壊的操作のため確認が必要です。",
        parameters=types.Schema(
            type=types.Type.OBJECT,
            properties={"slug": types.Schema(type=types.Type.STRING)},
            required=["slug"],
        ),
    ),
])
 
config = types.GenerateContentConfig(
    tools=[publish_tool],
    # 自動実行を止めて、呼び出し提案だけ受け取ります
    automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True),
)
 
contents = [
    types.Content(role="user", parts=[
        types.Part(text="レビュー済みの記事を公開してください。対象: gemini-batch-api-notes"),
    ]),
]
 
MAX_STEPS = 6  # 暴走防止。経験上、定型運用タスクは6ステップで収まります
 
for step in range(MAX_STEPS):
    response = client.models.generate_content(
        model="gemini-3.5-flash", contents=contents, config=config,
    )
    part = response.candidates[0].content.parts[0]
 
    if not part.function_call:
        print(response.text)  # ツール提案がなくなったら完了
        break
 
    call = part.function_call
    # ここで呼び出し内容を検査してから実行します
    result = execute_with_allowlist(call.name, dict(call.args))
 
    contents.append(response.candidates[0].content)
    contents.append(types.Content(role="user", parts=[
        types.Part.from_function_response(name=call.name, response={"result": result}),
    ]))

READ_TOOLS = {"get_article_stats", "list_pending_drafts"}
WRITE_TOOLS = {"publish_article"}
 
def execute_with_allowlist(name: str, args: dict) -> str:
    if name in READ_TOOLS:
        return run_tool(name, args)
    if name in WRITE_TOOLS:
        # 破壊系は対象の実在確認を挟んでから実行します
        if not draft_exists(args.get("slug", "")):
            return f"対象 {args.get('slug')} が見つからないため中止しました"
        return run_tool(name, args)
    return f"ツール {name} は許可されていません"