2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@har1101

Claude Agent SDKを使ってClaude Codeのようにユーザーと対話するための設定

2
Posted at

こんにちは、ふくちです。最近Claude Agent SDK使う機会があるので、気になったことを試してみました。

背景

Claude Agent SDKを使ってAIエージェントを構築する際、permission_mode="bypassPermissions"を設定してツールを自動許可にすることがあると思います。自動化パイプラインで人間の介入なしに動かしたいケースや、細やかな権限設定を別(クラウド側など)で行っている時などです。

一方で、エージェントが途中でユーザーに質問したい場面もあると思います。Claude CodeやClaude Desktopだとよく見るやつですね。

Claude Agent SDKは実質Claude Codeなので、もちろん実現可能です。
AskUserQuestionというビルトインツールがあり、can_use_toolコールバックを通じてユーザー入力を受け取ることができます。

そこで生じた疑問は、bypassPermissionsモードでAskUserQuestionのcan_use_toolコールバックは発火するのか？ということ。

ドキュメントには明示的な記載がなく、bypassPermissionsがツールの自動許可を意味するならcan_use_toolが呼ばれない（=ユーザーに質問が届かない）可能性があると思ったため、試してみました。

can_use_toolコールバックの仕組み

まず前提として、can_use_toolコールバックの仕組みを整理しておきます。

Claude Agent SDKでは、Claudeがツールを使おうとしたとき、SDKがツール実行の手前でインターセプトしてくれます。開発者は「ツールが使われそうになったらこの関数を呼んでね」とあらかじめSDKに登録しておく形です。ここで登録するのがcan_use_toolコールバック関数です。

# ① 開発者がコールバック関数を定義する
async def my_callback(tool_name, input_data, context):
    if tool_name == "Bash":
        print(f"Claudeが {input_data['command']} を実行しようとしています")
        # 許可 or 拒否を返す
    return PermissionResultAllow(updated_input=input_data)

# ② SDKに登録する
options = ClaudeAgentOptions(
    can_use_tool=my_callback,  # ← ここで登録
)

# ③ Claudeが動き出す
#    ツールを使おうとした瞬間、SDKが②で登録されたmy_callbackを呼ぶ
#    callbackが結果を返すまで、Claudeは一時停止する
async for message in query(prompt=..., options=options):
    ...

本来の用途は「Claudeがファイルを削除しようとしてるけど、本当にいい？」のようなツール使用の承認フローです。

AskUserQuestion: 承認の仕組みを流用した対話

ここでAskUserQuestionというビルトインツールがあります。Claudeが「ユーザーに質問したい」と判断したとき、このツールを呼びます。

するとSDKがcan_use_toolコールバックを呼び出し、コールバックの引数input_dataに質問内容が渡ってきます。開発者はコールバック内でユーザーに質問を表示し、回答をPermissionResultAllowで返せばOK。

つまり、ツール承認の仕組みをそのまま流用してユーザーとの対話を実現している形です。

bypassPermissionsとの矛盾

一方、permission_mode="bypassPermissions"という設定があります。自動化パイプラインなど人間の介入なしに動かしたいケースで、全ツールを自動許可にするモードです。

ここで矛盾が生じます。bypassPermissionsは「SDKがコールバックを呼ばず、全部自動で許可する」モード。しかしAskUserQuestionは「コールバックが呼ばれること」を前提にユーザー入力を受け取ります。

自動許可モードがAskUserQuestionのコールバックまでスキップしてしまったら、質問がユーザーに届きません。

ドキュメントにはこの組み合わせについて明示的な記載がないため、実際に動くのか検証しました。

なお、もう1点気になることがあります。SDKドキュメントのPython版サンプルには、can_use_toolを使う際にダミーのPreToolUseフックを登録する必要があると書かれています。これはストリームを開いたままにするためのワークアラウンドで、フックがないとツール実行前にストリームが閉じてしまい、コールバックが呼ばれないそうです。

# ドキュメントに記載されているワークアラウンド
async def dummy_hook(input_data, tool_use_id, context):
    return {"continue_": True}

options = ClaudeAgentOptions(
    can_use_tool=my_callback,
    hooks={
        "PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])],
    },
)

bypassPermissionsとの組み合わせでもこのダミーフックが必要なのかは不明だったので、フックあり/なしの2パターンで検証しています。

言葉だけだと分かり辛いので簡単な図にすると以下の形です。
まずはbypassPermissions無しで、普通に実行した場合。

ポイントは、どちらもcan_use_toolコールバックを経由することです。そしてClaudeはコールバックが返るまで一時停止します。

まずはbypassPermissionsを設定してエージェントを実行した場合。

普通のツール使用時はコールバック関数が呼ばれず即座にレスポンスが返されます。
ではAskUserQuestionの場合は？という疑問につながりました。期待値としては、bypassPermissionsが設定されていてもコールバック関数が発火することです。

実験コード

環境は以下の感じです。

  • claude-agent-sdk(Python版) 0.1.56
  • Python 3.13
  • uv 0.10.4
  • macOS

コード自体はClaude Codeに作ってもらいました。

#!/usr/bin/env python3
"""bypassPermissionsモードでAskUserQuestion + can_use_toolが動作するか検証する。"""

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, query, HookMatcher
from claude_agent_sdk.types import PermissionResultAllow, PermissionResultDeny


CALLBACK_FIRED = False
CALLBACK_TOOL_NAME = ""


async def can_use_tool(tool_name, input_data, context):
    global CALLBACK_FIRED, CALLBACK_TOOL_NAME
    CALLBACK_FIRED = True
    CALLBACK_TOOL_NAME = tool_name

    print(f"\n{'='*60}")
    print(f"✅ can_use_tool FIRED!")
    print(f"   tool_name: {tool_name}")
    print(f"   input_data keys: {list(input_data.keys()) if isinstance(input_data, dict) else type(input_data)}")

    if tool_name == "AskUserQuestion":
        questions = input_data.get("questions", [])
        print(f"   questions count: {len(questions)}")
        for i, q in enumerate(questions):
            print(f"   Q{i+1}: {q.get('question', '?')}")
            for opt in q.get("options", []):
                print(f"       - {opt.get('label')}: {opt.get('description', '')}")
        # 最初の選択肢で自動回答（テスト用）
        answers = {}
        for q in questions:
            first_option = q.get("options", [{}])[0].get("label", "Yes")
            answers[q["question"]] = first_option
        print(f"   Auto-answering: {answers}")
        return PermissionResultAllow(updated_input={"questions": questions, "answers": answers})
    else:
        print(f"   (Non-AskUserQuestion tool, auto-allowing)")
        return PermissionResultAllow(updated_input=input_data)


async def dummy_hook(input_data, tool_use_id, context):
    """SDKの要件: can_use_toolのためにストリームを開いたままにするダミーフック。"""
    return {"continue_": True}


async def prompt_stream():
    yield {
        "type": "user",
        "session_id": "",
        "message": {
            "role": "user",
            "content": (
                "ユーザーに以下の質問をしてください。AskUserQuestionツールを使って質問してください:\n"
                "「このプロジェクトで使用するデータベースはどれですか？」\n"
                "選択肢: PostgreSQL, MySQL, DynamoDB\n"
                "質問したら、ユーザーの回答を報告して終了してください。"
            ),
        },
        "parent_tool_use_id": None,
    }


async def run_test(test_name, options):
    """テストを実行してコールバック発火の有無を返す。"""
    global CALLBACK_FIRED, CALLBACK_TOOL_NAME
    CALLBACK_FIRED = False
    CALLBACK_TOOL_NAME = ""

    print(f"\n{'='*60}")
    print(f"{test_name}")
    print(f"{'='*60}")

    try:
        async for message in query(prompt=prompt_stream(), options=options):
            if hasattr(message, "content"):
                for block in message.content:
                    if hasattr(block, "text"):
                        print(f"[AssistantMessage] {block.text[:300]}")
            elif hasattr(message, "result"):
                print(f"[ResultMessage] session={message.session_id} turns={message.num_turns}")
    except Exception as e:
        print(f"[ERROR] {type(e).__name__}: {e}")

    print(f"\n--- Results ---")
    print(f"  can_use_tool fired: {CALLBACK_FIRED}")
    print(f"  tool_name: {CALLBACK_TOOL_NAME}")
    return CALLBACK_FIRED


async def main():
    print("claude-agent-sdk: AskUserQuestion + bypassPermissions検証")
    print("="*60)

    # Test 1: ダミーフック付き
    result1 = await run_test(
        "TEST 1: bypassPermissions + can_use_tool +ダミーフック付き",
        ClaudeAgentOptions(
            allowed_tools=["Read", "AskUserQuestion"],
            permission_mode="bypassPermissions",
            can_use_tool=can_use_tool,
            model="sonnet",
            max_turns=5,
            hooks={
                "PreToolUse": [HookMatcher(matcher=None, hooks=[dummy_hook])],
            },
        ),
    )

    # Test 2: ダミーフックなし
    result2 = await run_test(
        "TEST 2: bypassPermissions + can_use_tool（ダミーフックなし）",
        ClaudeAgentOptions(
            allowed_tools=["Read", "AskUserQuestion"],
            permission_mode="bypassPermissions",
            can_use_tool=can_use_tool,
            model="sonnet",
            max_turns=5,
        ),
    )

    print(f"\n{'='*60}")
    print("SUMMARY")
    print(f"{'='*60}")
    print(f"  Test 1 (ダミーフック付き): {'✅ OK' if result1 else '❌ NG'}")
    print(f"  Test 2 (ダミーフックなし): {'✅ OK' if result2 else '❌ NG'}")


if __name__ == "__main__":
    asyncio.run(main())

結果

Test1のAskUserQuestion Tool + bypassPermissions + ダミーフックあり は、正しく動作しました。

✅ can_use_tool FIRED!
   tool_name: AskUserQuestion
   input_data keys: ['questions']
   questions count: 1
   Q1: このプロジェクトで使用するデータベースはどれですか？
       - PostgreSQL: オープンソースのリレーショナルデータベース。高機能で拡張性が高い。
       - MySQL: 広く普及しているオープンソースのリレーショナルデータベース。
       - DynamoDB: AWSが提供するフルマネージドのNoSQLデータベース。スケーラビリティが高い。
   Auto-answering: {'このプロジェクトで使用するデータベースはどれですか？': 'PostgreSQL'}

[AssistantMessage] ユーザーの回答を報告します：
**質問**: このプロジェクトで使用するデータベースはどれですか？
**回答**: **PostgreSQL**
ユーザーは **PostgreSQL** を選択しました。以上で終了です。

[ResultMessage] session=9f272b8c-... turns=3
  • can_use_toolが正しく発火した
  • input_dataにquestions配列が含まれ、質問テキスト・選択肢が取得できた
  • 回答をPermissionResultAllowで返却すると、Claude Codeが処理を継続した
  • 全体のターン数=3（プロンプト→AskUserQuestion→回答レポート）

Test2のAskUserQuestion Tool + bypassPermissions + ダミーフックなしだと、Stream closedとして失敗しました。

[AssistantMessage] 申し訳ありません、ツールの実行中にエラーが発生しました。もう一度試みます。
[AssistantMessage] `AskUserQuestion` ツールの実行中に「Stream closed」エラーが発生し、
質問を表示できませんでした。

代わりに、テキストで質問させていただきます：

---
**このプロジェクトで使用するデータベースはどれですか？**
- **A) PostgreSQL** — オープンソースのリレーショナルDB。...
- **B) MySQL** — ...

[ResultMessage] session=a07bccbd-... turns=4
  • can_use_toolは発火しなかった
  • AskUserQuestionツール呼び出し時に"Stream closed"エラーが発生
  • Claude Codeはフォールバックとしてテキストで質問を出力した（＝ツールなしで動く場合と同じ挙動）
  • SDKドキュメントに記載されている、ダミーPreToolUseフックは必須

まとめ

最後にまとめると以下のとおりです。

条件 can_use_tool 発火 備考
bypassPermissions+can_use_tool+ダミーフックあり Yes 正常動作
bypassPermissions+can_use_tool+ダミーフックなし No Stream closedエラー

  1. bypassPermissionsとAskUserQuestionは共存可能
    →bypassPermissionsは他のツール（Bash, Write等）の許可確認をスキップするが、AskUserQuestionのcan_use_toolコールバックは正しく発火します

  2. ダミーPreToolUseフックは必須
    →これがないとcan_use_toolが呼ばれる前にストリームが閉じてしまいます(SDKドキュメントの記載通り)

実装サンプル

Claude Agent SDKでbypassPermissionsを使いつつ対話的なユーザー入力を実現するには:

async def keep_stream_open(input_data, tool_use_id, context):
    return {"continue_": True}

options = ClaudeAgentOptions(
    allowed_tools=[..., "AskUserQuestion"],
    permission_mode="bypassPermissions",
    can_use_tool=my_handler,
    hooks={
        "PreToolUse": [HookMatcher(matcher=None, hooks=[keep_stream_open])],
    },
)

この3点セット（AskUserQuestion in allowed_tools + can_use_tool +ダミーフック）が最小構成となるようです。

で、これをAgentCore Runtime上でやるならどーすんねんという話になっていきそうです。またそこはちょっと色々試してみます。

参考リンク

2
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?