지연 툴(Deferred Tools)

지연 툴(Deferred Tools)

모델이 같은 에이전트 실행·같은 Python 프로세스 동안 실행되어서는 안 되거나 실행될 수 없는 툴을 호출해야 하는 몇 가지 시나리오가 있어요:

  • 먼저 사용자 승인이 필요할 수 있어요.
  • 업스트림 서비스, 프론트엔드, 또는 사용자가 결과를 제공하기를 의존할 수 있어요.
  • 결과 생성이 에이전트 프로세스를 실행 상태로 유지하는 게 합리적인 것보다 오래 걸릴 수 있어요.

이러한 사용 사례를 지원하기 위해 Pydantic AI는 지연 툴(deferred tools)의 개념을 제공하며, 아래에 문서화된 두 가지 형태가 있어요:

모델이 지연 툴을 호출하면 두 가지 방식으로 해결할 수 있어요:

  • 인라인으로 해결 — 대기 중인 호출의 일부 또는 전부를 해결하는 핸들러를 가진 HandleDeferredToolCalls 기능을 사용해요. 에이전트 실행은 끝나고 재시작할 필요 없이 단일 호출로 계속돼요. 해결자(예: 승인 게이트, 외부 서비스 클라이언트)가 에이전트와 같은 프로세스에 있을 때 사용하세요. 핸들러로 지연 호출 해결 참고.
  • 실행을 끝내기 — 지연 툴 호출에 대한 정보가 담긴 DeferredToolRequests 출력 객체로 실행을 끝내요. 호출자는 승인/결과를 모은 뒤 원래 실행의 메시지 이력DeferredToolResults 객체로 새 에이전트 실행을 시작해요. 그 후속은 자체 run_id를 가진 별도의 에이전트 실행이에요(일시정지된 실행을 재사용하지 마세요). conversation_id로 일시정지/재개 연관을 유지하세요. 해결자가 에이전트 프로세스 밖에 있을 때 사용하세요. 예를 들어 대기 중인 호출을 사용자에게 드러내고 응답을 받으면 후속 실행을 시작하는 UI 어댑터요.

두 흐름은 합성돼요. 핸들러가 호출의 일부를 해결하고 나머지는 바깥 호출자가 처리하도록 DeferredToolRequests 출력으로 버블링할 수 있어요.

stop-the-world 흐름은 에이전트 실행 출력의 가능한 타입이 올바르게 추론되도록 DeferredToolRequestsAgentoutput_type에 있어야 해요. 에이전트가 지연 툴을 사용할 수 없는 컨텍스트에서도 쓰이고 그 타입을 에이전트를 쓰는 곳마다 다루고 싶지 않다면, 대신 agent.run(), agent.run_sync(), agent.run_stream(), agent.iter()로 에이전트를 실행할 때 output_type 인자를 전달할 수 있어요. 실행 시 output_type은 생성 시 지정한 것을(타입 추론 이유로) 덮어쓰므로 원래 출력 타입을 명시적으로 포함해야 해요.

출처: 문서

본문

핸들러로 지연 호출 해결

지연 툴 호출을 처리하는 권장 방법은 DeferredToolRequests를 받고 일부 또는 전부를 해결하는 DeferredToolResults를 반환하는 핸들러를 가진 HandleDeferredToolCalls 기능을 등록하는 것이에요. 툴 실행 파이프라인이 결과를 인라인으로 적용하고 에이전트 실행은 지연 툴이 정상적으로 반환한 것처럼 단일 호출로 계속돼요.

핸들러가 있으면 DeferredToolRequests를 더 이상 출력 타입으로 선언할 필요가 없어요. 해결되지 않은 호출을 호출자에게 버블링하고 싶지 않다면요(아래 참고).

DeferredToolRequests.build_results()는 편의 생성자예요. 모든 툴 호출 ID가 올바른 종류의 대기 요청을 가리키는지 검증하고, 별도로 지정되지 않은 승인 요청을 자동 승인하는 approve_all=True를 받아요.

from pydantic_ai import (
    Agent,
    ApprovalRequired,
    CallDeferred,
    DeferredToolRequests,
    DeferredToolResults,
    RunContext,
    ToolDenied,
)
from pydantic_ai.capabilities import HandleDeferredToolCalls


async def handle_deferred(
    ctx: RunContext, requests: DeferredToolRequests
) -> DeferredToolResults:
    approvals: dict[str, bool | ToolDenied] = {}
    for call in requests.approvals:
        if call.tool_name == 'delete_file':
            approvals[call.tool_call_id] = ToolDenied('Deleting files is not allowed')
        else:
            approvals[call.tool_call_id] = True

    calls = {call.tool_call_id: f'(external result for {call.tool_name})' for call in requests.calls}

    return requests.build_results(approvals=approvals, calls=calls)


agent = Agent(
    'openai:gpt-5.2',
    capabilities=[HandleDeferredToolCalls(handler=handle_deferred)],
)


@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    return f'File {path!r} deleted'  # (1)


@agent.tool
def update_file(ctx: RunContext, path: str, content: str) -> str:
    if path == '.env' and not ctx.tool_call_approved:
        raise ApprovalRequired
    return f'File {path!r} updated: {content!r}'


@agent.tool_plain
async def send_to_worker(task: str) -> str:
    raise CallDeferred  # (2)

(1) 여기 절대 도달하지 않아요 — 핸들러가 이 호출을 거부하므로 모델은 대신 거부 메시지를 봐요.

(2) 핸들러가 이 외부 호출의 결과를 제공하므로 툴 본문은 지연을 신호하기만 해요.

핸들러가 호출의 일부 또는 전부를 해결하지 않기로 하면(반환된 DeferredToolResults에서 생략하거나 None 반환), 다음 HandleDeferredToolCalls(또는 handle_deferred_tool_calls 훅을 덮어쓰는 다른 어떤 기능)가 기회를 얻고, 여전히 해결되지 않은 호출은 DeferredToolRequests 출력으로 버블링돼요. 그 버블링을 허용하려면 DeferredToolRequests를 에이전트의 output_type에 포함하세요. 그러면 말이 될 때 인라인 처리를 stop-the-world 흐름과 결합할 수 있어요.

커스텀 기능을 구축 중이고 승인이나 외부 호출을 스스로 해결해야 한다면(예: 지연 툴을 노출하는 샌드박스), 별도 HandleDeferredToolCalls를 등록하는 대신 기능에서 handle_deferred_tool_calls 훅을 직접 덮어쓰세요. 같은 훅은 Hooks 기능을 통해서도 사용할 수 있어요. Hooks 참고.

아래 섹션들은 핸들러가 해결할 수 있는 두 가지 지연 툴과 각각의 대안 stop-the-world 흐름을 묘사해요. 여러 기능이 어떻게 합성되는지는 WrapperCapabilitycapabilities=[...] 목록을 포함해 Capabilities를 참고하세요.

인간-인-루프 툴 승인

툴 함수가 항상 승인을 요구한다면, @agent.tool 데코레이터, @agent.tool_plain 데코레이터, Tool 클래스, FunctionToolset.tool 데코레이터, FunctionToolset.add_function() 메서드에 requires_approval=True 인자를 전달할 수 있어요. 함수 안에서 툴 호출이 승인됐다고 가정할 수 있어요.

realtime 세션에서는 승인을 인라인으로 해결해야 하며, 보통 HandleDeferredToolCalls 핸들러로 해요(기능 훅도 해결할 수 있어요). 아무것도 해결하지 않는 호출은 매번 거부돼요.

승인이 툴 호출의 인자나 에이전트 실행 컨텍스트(의존성이나 메시지 이력 같은)에 의존한다면, 툴 함수에서 ApprovalRequired를 발생시키세요. 툴 호출이 이미 승인됐으면 RunContext.tool_call_approved 속성이 True가 돼요.

그것을 툴의 args_validator에서도 발생시킬 수 있어요. 그 툴 함수 전에 실행되어 사람에게 승인을 요청하기 전에 잘못된 인자를 거부하게 해요.

toolset(MCP 서버 같은)이 제공하는 툴 호출에 승인을 요구하려면 ApprovalRequiredToolset 문서를 참고하세요.

승인은 신뢰할 수 없는 클라이언트에 대한 인가 경계가 아니다

UI 어댑터로 에이전트를 서빙할 때, 승인 결정은 나머지 요청과 함께 클라이언트가 제출해요. 어댑터는 자신이 발행한 툴 호출에 대한 서버 측 기록이 없어요. 엔드포인트에 닿을 수 있는 클라이언트는 자신이 만든 툴 호출을 승인할 수 있어요. 인간-인-루프 승인은 인간의 서명 없이 모델 이 행동하는 것을 보호해요. 어댑터 엔드포인트를 인증하고 툴 함수 자체 안에서 민감한 행동에 대한 인가를 시행하는 것을 대체하지 않아요. 함수는 호출이 이력에 어떻게 들어왔든 실행되니까요. 이것은 클라이언트의 message_history를 받아들이는 어떤 엔드포인트에도 적용돼요. 어댑터뿐만 아니라요. 클라이언트 제공 이력에 대한 신뢰 경계클라이언트 제출 메시지에 대한 신뢰 모델 참고.

모델이 승인을 요구하는 툴을 호출하면 에이전트 실행은 툴 이름, 검증된 인자, 고유 툴 호출 ID를 담은 ToolCallPart들로 된 approvals 목록을 가진 DeferredToolRequests 출력 객체로 끝나요.

사용자의 승인이나 거부를 모은 뒤, DeferredToolResults 객체를 만들 수 있어요. approvals 사전은 각 툴 호출 ID를 boolean, ToolApproved 객체(선택적 override_args 포함), 또는 ToolDenied 객체(모델에 줄 선택적 커스텀 message 포함)에 매핑해요. DeferredToolResults에 각 툴 호출 ID를 툴의 RunContext.tool_call_metadata 속성에서 사용할 수 있는 메타데이터 사전에 매핑하는 metadata 사전을 제공할 수도 있어요. 이 DeferredToolResults 객체는 원래 실행의 메시지 이력과 함께 에이전트 실행 메서드 중 하나에 deferred_tool_results로 제공할 수 있어요.

모든 파일 삭제와 특정 보호 파일의 업데이트에 승인을 요구하는 예시:

from pydantic_ai import (
    Agent,
    ApprovalRequired,
    DeferredToolRequests,
    DeferredToolResults,
    RunContext,
    ToolDenied,
)

agent = Agent('openai:gpt-5.2', output_type=[str, DeferredToolRequests])

PROTECTED_FILES = {'.env'}


@agent.tool
def update_file(ctx: RunContext, path: str, content: str) -> str:
    if path in PROTECTED_FILES and not ctx.tool_call_approved:
        raise ApprovalRequired(metadata={'reason': 'protected'})  # (1)
    return f'File {path!r} updated: {content!r}'


@agent.tool_plain(requires_approval=True)
def delete_file(path: str) -> str:
    return f'File {path!r} deleted'


result = agent.run_sync('Delete `__init__.py`, write `Hello, world!` to `README.md`, and clear `.env`')
messages = result.all_messages()

assert isinstance(result.output, DeferredToolRequests)
requests = result.output
print(requests)
"""
DeferredToolRequests(
    calls=[],
    approvals=[
        ToolCallPart(
            tool_name='update_file',
            args={'path': '.env', 'content': ''},
            tool_call_id='update_file_dotenv',
        ),
        ToolCallPart(
            tool_name='delete_file',
            args={'path': '__init__.py'},
            tool_call_id='delete_file',
        ),
    ],
    metadata={'update_file_dotenv': {'reason': 'protected'}},
)
"""

results = DeferredToolResults()
for call in requests.approvals:
    result = False
    if call.tool_name == 'update_file':
        # Approve all updates
        result = True
    elif call.tool_name == 'delete_file':
        # deny all deletes
        result = ToolDenied('Deleting files is not allowed')

    results.approvals[call.tool_call_id] = result

result = agent.run_sync(
    'Now create a backup of README.md',  # (2)
    message_history=messages,
    deferred_tool_results=results,
)
print(result.output)
"""
Here's what I've done:
- Attempted to delete __init__.py, but deletion is not allowed.
- Updated README.md with: Hello, world!
- Cleared .env (set to empty).
- Created a backup at README.md.bak containing: Hello, world!

If you want a different backup name or format (e.g., timestamped like README_2025-11-24.bak), let me know.
"""
print(result.all_messages())
"""
[
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Delete `__init__.py`, write `Hello, world!` to `README.md`, and clear `.env`',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='delete_file',
                args={'path': '__init__.py'},
                tool_call_id='delete_file',
            ),
            ToolCallPart(
                tool_name='update_file',
                args={'path': 'README.md', 'content': 'Hello, world!'},
                tool_call_id='update_file_readme',
            ),
            ToolCallPart(
                tool_name='update_file',
                args={'path': '.env', 'content': ''},
                tool_call_id='update_file_dotenv',
            ),
        ],
        usage=RequestUsage(
            cost=Decimal('0.00040425'), input_tokens=63, output_tokens=21
        ),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='update_file',
                content="File 'README.md' updated: 'Hello, world!'",
                tool_call_id='update_file_readme',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='delete_file',
                content='Deleting files is not allowed',
                tool_call_id='delete_file',
                timestamp=datetime.datetime(...),
                outcome='denied',
            ),
            ToolReturnPart(
                tool_name='update_file',
                content="File '.env' updated: ''",
                tool_call_id='update_file_dotenv',
                timestamp=datetime.datetime(...),
            ),
            UserPromptPart(
                content='Now create a backup of README.md',
                timestamp=datetime.datetime(...),
            ),
        ],
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='update_file',
                args={'path': 'README.md.bak', 'content': 'Hello, world!'},
                tool_call_id='update_file_backup',
            )
        ],
        usage=RequestUsage(
            cost=Decimal('0.0005845'), input_tokens=86, output_tokens=31
        ),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='update_file',
                content="File 'README.md.bak' updated: 'Hello, world!'",
                tool_call_id='update_file_backup',
                timestamp=datetime.datetime(...),
            )
        ],
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content="Here's what I've done:\n- Attempted to delete __init__.py, but deletion is not allowed.\n- Updated README.md with: Hello, world!\n- Cleared .env (set to empty).\n- Created a backup at README.md.bak containing: Hello, world!\n\nIf you want a different backup name or format (e.g., timestamped like README_2025-11-24.bak), let me know."
            )
        ],
        usage=RequestUsage(
            cost=Decimal('0.00140875'), input_tokens=93, output_tokens=89
        ),
        model_name='gpt-5.2',
        timestamp=datetime.datetime(...),
        run_id='...',
        conversation_id='...',
    ),
]
"""

선택적 metadata 매개변수는 tool_call_id로 키가 지정된 DeferredToolRequests.metadata에서 접근할 수 있는 임의 컨텍스트를 지연 툴 호출에 붙일 수 있어요.

이 두 번째 에이전트 실행은 첫 번째가 멈춘 곳에서 계속되고, 툴 승인 결과와 선택적으로 새 user_prompt를 제공해 지연 결과와 함께 모델에 추가 지시를 줘요.

(이 예제는 완전해서 "그대로" 실행할 수 있어요)

툴 결과 순서

툴 결과는 모델이 해당 툴 호출을 방출한 순서를 따라요. 위 메시지 이력에서 delete_file의 거부 결과가 .env에 대한 update_file의 결과 앞에 나타나는데, 모델이 delete_file을 먼저 방출했기 때문이에요. 이것은 v2의 의도적인 동작 변경이에요. 결과는 더 이상 툴 종류별로 그룹화되지 않아서, 보이는 순서가 모델의 방출 순서를 반영해요.

외부 툴 실행

툴 호출의 결과를 호출된 같은 에이전트 실행 안에서 생성할 수 없으면 그 툴은 외부로 간주돼요. 외부 툴의 예는 웹이나 앱 프론트엔드가 구현하는 클라이언트 측 툴, 그리고 에이전트 프로세스를 실행 상태로 두는 대신 백그라운드 워커나 외부 서비스에 넘겨지는 느린 태스크예요.

툴 호출을 외부에서 실행할지 여부가 툴 호출 인자, 에이전트 실행 컨텍스트(의존성이나 메시지 이력 같은), 수행할 작업이 걸릴 시간에 의존하면, 툴 함수를 정의하고 조건부로 CallDeferred 예외를 발생시킬 수 있어요. 예외를 발생시키기 전에 툴 함수는 보통 어떤 백그라운드 태스크를 예약하고 RunContext.tool_call_id를 함께 넘겨 나중에 결과를 지연 툴 호출과 매칭할 수 있게 해요.

승인처럼 툴의 args_validatorCallDeferred를 발생시킬 수 있으므로, 유효한 인자를 가진 호출만 넘겨져요.

툴이 항상 외부에서 실행되고 그 정의가 인자의 JSON 스키마와 함께 코드에 제공된다면 ExternalToolset을 사용할 수 있어요. 외부 툴을 사전에 알고 인자 JSON 스키마가 준비되어 있지 않다면, 적절한 시그니처로 CallDeferred 예외만 발생시키는 툴 함수를 정의할 수도 있어요.

모델이 외부 툴을 호출하면 에이전트 실행은 툴 이름, 검증된 인자, 고유 툴 호출 ID를 담은 ToolCallPart들로 된 calls 목록을 가진 DeferredToolRequests 출력 객체로 끝나요.

툴 호출 결과가 준비되면 DeferredToolResults 객체를 만들 수 있어요. calls 사전은 각 툴 호출 ID를 모델에 반환할 임의 값, ToolReturn 객체, 또는 툴 호출이 실패한 경우의 예외([모델이 다시 시도해야 하면 ModelRetry, 실패를 실패 결과로 보고해야 하면 툴의 재시도 예산을 소비하지 않고 어떻게 진행할지 결정하게 하는 ToolFailed)에 매핑해요. 이 DeferredToolResults 객체는 원래 실행의 메시지 이력과 함께 에이전트 실행 메서드 중 하나에 deferred_tool_results로 제공할 수 있어요.

완료하는 데 시간이 걸리는 태스크를 백그라운드로 옮기고 완료되면 결과를 모델에 반환하는 예시:

import asyncio
from dataclasses import dataclass
from typing import Any

from pydantic_ai import (
    Agent,
    CallDeferred,
    DeferredToolRequests,
    DeferredToolResults,
    ModelRetry,
    RunContext,
)


@dataclass
class TaskResult:
    task_id: str
    result: Any


async def calculate_answer_task(task_id: str, question: str) -> TaskResult:
    await asyncio.sleep(1)
    return TaskResult(task_id=task_id, result=42)


agent = Agent('openai:gpt-5.2', output_type=[str, DeferredToolRequests])

tasks: list[asyncio.Task[TaskResult]] = []


@agent.tool
async def calculate_answer(ctx: RunContext, question: str) -> str:
    task_id = f'task_{len(tasks)}'  # (1)
    task = asyncio.create_task(calculate_answer_task(task_id, question))
    tasks.append(task)

    raise CallDeferred(metadata={'task_id': task_id})  # (2)


async def main():
    result = await agent.run('Calculate the answer to the ultimate question of life, the universe, and everything')
    messages = result.all_messages()

    assert isinstance(result.output, DeferredToolRequests)
    requests = result.output
    print(requests)
    """
    DeferredToolRequests(
        calls=[
            ToolCallPart(
                tool_name='calculate_answer',
                args={
                    'question': 'the ultimate question of life, the universe, and everything'
                },
                tool_call_id='pyd_ai_tool_call_id',
            )
        ],
        approvals=[],
        metadata={'pyd_ai_tool_call_id': {'task_id': 'task_0'}},
    )
    """

    done, _ = await asyncio.wait(tasks)  # (3)
    task_results = [task.result() for task in done]
    task_results_by_task_id = {result.task_id: result.result for result in task_results}

    results = DeferredToolResults()
    for call in requests.calls:
        try:
            task_id = requests.metadata[call.tool_call_id]['task_id']
            result = task_results_by_task_id[task_id]
        except KeyError:
            result = ModelRetry('No result for this tool call was found.')

        results.calls[call.tool_call_id] = result

    result = await agent.run(message_history=messages, deferred_tool_results=results)
    print(result.output)
    #> The answer to the ultimate question of life, the universe, and everything is 42.
    print(result.all_messages())
    """
    [
        ModelRequest(
            parts=[
                UserPromptPart(
                    content='Calculate the answer to the ultimate question of life, the universe, and everything',
                    timestamp=datetime.datetime(...),
                )
            ],
            timestamp=datetime.datetime(...),
            run_id='...',
            conversation_id='...',
        ),
        ModelResponse(
            parts=[
                ToolCallPart(
                    tool_name='calculate_answer',
                    args={
                        'question': 'the ultimate question of life, the universe, and everything'
                    },
                    tool_call_id='pyd_ai_tool_call_id',
                )
            ],
            usage=RequestUsage(
                cost=Decimal('0.00029225'), input_tokens=63, output_tokens=13
            ),
            model_name='gpt-5.2',
            timestamp=datetime.datetime(...),
            run_id='...',
            conversation_id='...',
        ),
        ModelRequest(
            parts=[
                ToolReturnPart(
                    tool_name='calculate_answer',
                    content=42,
                    tool_call_id='pyd_ai_tool_call_id',
                    timestamp=datetime.datetime(...),
                )
            ],
            timestamp=datetime.datetime(...),
            run_id='...',
            conversation_id='...',
        ),
        ModelResponse(
            parts=[
                TextPart(
                    content='The answer to the ultimate question of life, the universe, and everything is 42.'
                )
            ],
            usage=RequestUsage(
                cost=Decimal('0.000504'), input_tokens=64, output_tokens=28
            ),
            model_name='gpt-5.2',
            timestamp=datetime.datetime(...),
            run_id='...',
            conversation_id='...',
        ),
    ]
    """

(1) 툴 호출 ID와 독립적으로 추적할 수 있는 태스크 ID를 생성해요.

(2) 선택적 metadata 매개변수가 task_id를 전달해 나중에 결과와 매칭할 수 있게 하며, tool_call_id로 키가 지정된 DeferredToolRequests.metadata에서 접근할 수 있어요.

(3) 현실에서는 보통 별도 프로세스에서 태스크 상태를 폴링하거나 모든 대기 태스크가 완료되면 알림을 받아요.

(이 예제를 실행하려면 asyncio를 import하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

스트림에서 지연 툴 호출 관찰

다른 어떤 툴 호출처럼, 지연 툴 호출은 이벤트 스트림FunctionToolCallEvent를 방출해요. 하지만 그 이벤트만으로는 스트림 소비자에게 호출이 상호작용을 기다리며 일시정지됐는지, 어떤 종류의 상호작용이 기대되는지 말해주지 않아요. 두 가지 추가 AgentStreamEvent가 그 컨텍스트를 담아요:

  • DeferredToolRequestsEvent — 지연 호출 배치당 한 번 방출되며 DeferredToolRequests를 담아요. 어떤 HandleDeferredToolCalls 핸들러가 실행되기 전에 방출되어, 소비자가 예를 들어 핸들러가 기다리는 동안 입력이 필요하다고 프런트엔드에 알릴 수 있어요. 핸들러가 모든 요청을 해결하지 않으면 실행은 대기 요청을 DeferredToolRequests 출력으로 끝나요.
  • DeferredToolResultsEvent — 핸들러가 요청의 (일부를) 해결할 때 방출되며 DeferredToolResults를 담아요. 해결된 호출은 정규 파이프라인을 통해 실행되어 각각 FunctionToolResultEvent를 방출해요. 결과가 그 대신 deferred_tool_results로 새 실행에 제공되면 이벤트가 방출되지 않아요. 그 경우 호출자가 이미 그것을 알기 때문이에요.

이것은 해결과 표현을 분리해요. 핸들러는 순수 해결 로직만 담을 수 있고(예: 지속 실행 워크플로우에서 신호 기다리기), 스트림 소비자는 어떤 툴이 인터랙티브인지 자체 매핑을 유지하지 않고 프런트엔드와의 모든 통신을 소유해요.

핸들러 예제를 계속하면:

from pydantic_ai import DeferredToolRequestsEvent, DeferredToolResultsEvent

from deferred_tool_handler import agent


async def main():
    async with agent.run_stream_events(
        'Delete `__init__.py`, write `Hello, world!` to `README.md`, and clear `.env`'
    ) as events:
        async for event in events:
            if isinstance(event, DeferredToolRequestsEvent):
                print(f'Approvals needed: {[call.tool_name for call in event.requests.approvals]}')
                #> Approvals needed: ['update_file', 'delete_file']
            elif isinstance(event, DeferredToolResultsEvent):
                print(f'Resolved: {list(event.results.approvals)}')
                #> Resolved: ['update_file_dotenv', 'delete_file']

(이 예제를 실행하려면 asyncio를 import하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

같이 보기

더 알아보기 (Learn more)