툴
툴
에이전트에 등록된 툴은 realtime 모델에 제공되고 백엔드에서 실행돼요. 세션은 인자를 검증하고, 재시도를 적용하고, 툴을 동시에 실행하며, 결과를 프로바이더에 반환하고, 나중에 핸드오프를 위해 평범한 툴 호출 메시지를 기록해요. 툴 호출 주변의 기능 훅은 기능과 훅에서 다뤄요.
출처: 문서
본문
함수 툴
모델이 툴을 호출하면 세션은 FunctionToolCallEvent를 방출하고, 툴을 실행하며, 결과를 반환하고, FunctionToolResultEvent를 방출해요. 파싱 실패와 ModelRetry는 표준 에이전트 실행처럼 RetryPromptPart를 만들어요. 다른 툴 예외는 세션을 끝내고 이터레이션에서 전파돼요. 이벤트 스트림을 전혀 반복하지 않았다면 그것은 오디오와 전사 뷰를 끝내고 세션이 닫힐 때 발생해요.(반복을 시작한 뒤 멈춘 소비자는 듣기를 멈추기로 선택한 것이므로, 그것을 대신해 발생하는 것은 없어요.) 실패한 호출은 outcome='failed'로 기록돼서, 확정된 이력을 Agent.run(message_history=...)에 전달할 수 있어요. 일반적인 on_tool_execute_error 기능 훅도 realtime에서 적용되고 예외를 대체 결과나 ModelRetry로 바꿔 모델이 회복할 수 있게 해요.
툴 반환 값은 표준 실행에서와 정확히 모델에 도달해요. 모델은 반환 값의 문자열 렌더링을 받아요. 프로바이더가 지원하는 곳에서는 ToolReturn의 content로 붙은 멀티모달 콘텐츠도 받아요. 로컬 이력은 return_value, content, metadata가 있는 전체 구조화된 ToolReturnPart를 유지해요. 붙은 콘텐츠는 실제로 전달되거나 크게 거부돼요. 조용히 열화되지 않아요. OpenAI와 Azure OpenAI는 텍스트와 이미지를 후속 사용자 메시지로 전달하고, Gemini Live의 툴 결과는 JSON 전용이라 텍스트는 결과에 접히고 어떤 바이너리 첨부든 UserError를 발생시켜요(#7362). 프로바이더가 담을 수 없는 미디어(어디서나 오디오와 문서, xAI에서 이미지도)는 마찬가지로 아무것도 보내기 전에 예외를 발생시켜요. 프로바이더가 진행 중인 호출을 취소하면 Pydantic AI는 태스크를 취소하고 로컬에 합성 취소 결과를 기록하며 그 결과를 프로바이더에 다시 보내지 않아요.
동시 툴 실행
모든 툴은 백그라운드에서 실행돼서, 느린 툴이 세션 이벤트, 다른 툴, 턴 추적을 막지 않아요. all_messages()는 호출이 순서 없이 끝나도 각 결과를 그 호출 옆에 유지해요.
모델이 기다리는 동안 계속 말하는지는 프로바이더별이에요. supports_async_tool_calls 프로필 플래그를 검사하세요. OpenAI와 Azure 모델은 일반적으로 그 간격을 채우고, Gemini는 지원되는 모델에서 google_async_tool_calls 설정(Live API에 툴을 NON_BLOCKING으로 선언)이 활성화되지 않으면 멈춰요.
네이티브 툴
프로바이더 네이티브 툴은 서버 측에서 실행돼요. WebSearch와 WebFetch 같은 하이레벨 기능이나 NativeTool로 추가하세요. 각 모델의 supported_native_tools 프로필이 진실의 원천이에요.
from pydantic_ai import Agent
from pydantic_ai.capabilities import WebSearch
from pydantic_ai.messages import NativeToolReturnPart, PartEndEvent
from pydantic_ai.realtime import RealtimeTurnCompleteEvent
agent = Agent(instructions='Answer questions, searching the web when useful.')
async def main():
async with agent.realtime(
'google:gemini-2.5-flash-native-audio-latest',
capabilities=[WebSearch()],
).session() as session:
await session.send("What's the latest Pydantic AI release?")
async for event in session:
if isinstance(event, PartEndEvent) and isinstance(event.part, NativeToolReturnPart):
print(event.part.content)
if isinstance(event, RealtimeTurnCompleteEvent):
break # keep listening in a real call; we stop after one reply
구성된 로컬 폴백이 있는 지원되지 않는 네이티브 툴은 연결 전에 대체돼요. 폴백이 없으면 세션을 열 때 UserError가 발생해요. 프로바이더·모델별 조합(Gemini 그라운딩, URL 컨텍스트, 함수 툴 제한 포함)은 Gemini 프로바이더 페이지가 표준이에요.
지연 및 승인 필요 툴
승인 게이트 툴은 HandleDeferredToolCalls 핸들러가 필요해요. 없으면 호출이 매번 거부돼요. 표준 실행은 DeferredToolRequests 출력으로 끝나고 사람이 답하면 재개할 수 있어요(지연 툴 참고). 하지만 라이브 대화는 멈출 곳이 없어요. 핸들러가 없으면 모델은 툴이 realtime 세션 중에 완료될 수 없다고 듣고 툴은 절대 실행되지 않아요.
핸들러는 각 호출을 인라인으로 해결해요. 승인(툴이 정상적으로 실행·반환), 거부(outcome='denied'로 기록), 결과 대체, 재시도 요청. 이 핸들러는 정책에서 작은 환불을 승인하고 나머지를 거부해요:
from pydantic_ai import Agent, DeferredToolRequests, DeferredToolResults, ToolDenied
from pydantic_ai.capabilities import HandleDeferredToolCalls
from pydantic_ai.tools import RunContext
agent = Agent(instructions='You are a customer support voice assistant.')
@agent.tool_plain(requires_approval=True)
def issue_refund(order_id: str, amount: float) -> str:
return f'Refunded ${amount:.2f} for order {order_id}.'
async def refund_policy(
ctx: RunContext[None], requests: DeferredToolRequests
) -> DeferredToolResults:
results = DeferredToolResults()
for call in requests.approvals:
if call.args_as_dict().get('amount', 0) <= 100:
results.approvals[call.tool_call_id] = True
else:
results.approvals[call.tool_call_id] = ToolDenied(
'Refunds over $100 need a human; offer to connect one.'
)
return results
async def main():
async with agent.realtime(
'openai:gpt-realtime',
capabilities=[HandleDeferredToolCalls(handler=refund_policy)],
).session():
...
이것은 호출이 지연되는 두 방식 모두에 적용돼요. 툴에서 ApprovalRequired나 CallDeferred를 발생시키는 것과, requires_approval=True나 외부 툴셋으로 미리 선언하는 것이요. 승인 게이트 툴은 표준 실행에서처럼 여전히 모델에 광고돼요. 그것을 호출하는 것은 툴을 실행하는 게 아니라 승인 흐름을 여는 거예요.
핸들러는 사람이 아니라 정책에서 답한다
핸들러는 신속히 결정을 반환해야 해요. 프로그래매틱 정책 해석자이지 승인 UI가 아니에요. 툴 자체처럼 백그라운드 태스크로 실행되어 세션의 이벤트를 절대 막지 않지만, 대화가 생각하는 동안 무엇을 하는지는 동시 툴 실행이 묘사하는 방식 그대로 프로바이더별이에요. OpenAI와 Azure는 계속하고, Gemini는 결과가 도착할 때까지 모델의 턴을 붙잡아요. 그래서 Gemini에서는 느린 핸들러가 어시스턴트 침묵으로 읽히고, 사용자가 그 간격에 말하면 프로바이더가 대기 중인 호출을 아예 취소해요(합성 취소로 기록).
호출 중간에 사람에게 묻고 답에서 재개하는 것은 아직 지원되지 않아요. realtime 세션은 멈추고 대역 외 결과를 위해 DeferredToolRequests 출력을 반환할 수 없어요. 호출 중에 요청을 해결하거나 그 워크플로우를 표준 에이전트 실행으로 옮기세요.
세션의 DeferredToolRequestsEvent는 같은 이유로 정보용이에요. 핸들러가 호출을 해결한 후에 방출되므로 소비자가 무엇이 요청되고 결정됐는지 관찰할 수 있어요. 그것은 응답할 훅이 아니에요. 표준 실행에서의 같은 이벤트와 달리 소비자를 기다리는 것이 없고, 핸들러가 설치되지 않고 호출이 거부됐을 때는 이벤트가 방출되지 않아요.
defer_loading=True로 등록된 툴은 관련된 이유로 realtime 세션에서 거부돼요. 지연 기능 로딩 참고.
프롬프트 큐잉
RunContext.enqueue() — 표준 실행에서 툴에서 후속 메시지 주입과 같은 메커니즘 — 은 realtime 툴이 텍스트나 SystemPromptPart를 큐에 넣게 해줘요. 세션을 구동하는 코드는 RealtimeSession.enqueue()를 직접 사용할 수 있어요. 예를 들어 대역 외 워치독 인스트럭션을 전달하듯이요:
전달은 세션의 이벤트 스트림에서 EnqueuedMessagesEvent로 보고돼요. 표준 실행과 일치해요.
import asyncio
from pydantic_ai import Agent
from pydantic_ai.messages import SystemPromptPart
from pydantic_ai.realtime import RealtimeTurnCompleteEvent
agent = Agent()
async def main():
async with agent.realtime('openai:gpt-realtime').session() as session:
session.enqueue(
SystemPromptPart(content='A watchdog detected elevated latency. Mention this briefly.'),
priority='when_idle',
)
async for event in session:
if isinstance(event, RealtimeTurnCompleteEvent):
break
if __name__ == '__main__':
asyncio.run(main())
기본 priority='asap'는 활성 응답이 끝난 후에 전달하고, priority='when_idle'은 모든 'asap' 항목 후에 모델이 유휴할 때까지 기다려요. 어느 우선순위도 어시스턴트 음성을 중단하지 않아요. 텍스트 파트와 시스템 파트는 하나의 사용자 턴으로 결합되고, 시스템 파트는 말하는 사람과 구별되도록 <system>...</system>으로 감싸져요. 시스템 파트는 텍스트가 어디서 왔는지 표시하지, 조용히 처리되어야 한다는 뜻은 아니에요. 모델은 여전히 턴을 얻고 응답하거나, 툴을 호출하거나, 넘어갈 수 있어요. 턴을 유도하지 않고 컨텍스트를 추가하려면 대신 send(..., respond=False)를 사용하세요. 전달된 턴은 실행 중간 메시지 주입에서처럼 이력에서 평범한 UserPromptPart가 돼요.
enqueue()는 send()를 대체하지 않아요. 진행 중인 응답이 끝나길 기다리는 반면, send()는 툴 호출이나 응답이 진행 중이어도 즉시 전달해요. 차례를 기다려야 하는 후속에는 enqueue(), 간격에 끼어들려면 send()를 잡으세요.
멀티모달 콘텐츠와 모델 응답은 realtime 라이브 입력 채널이 그 표준 실행 시맨틱을 보존할 수 없으므로 거부돼요.
툴에서 세션 끝내기
툴에서 끊으려면 ctx.realtime_session을 통해 close()를 호출하세요:
from pydantic_ai import Agent, RunContext
agent = Agent(instructions='When the caller says goodbye, call `hang_up`.')
@agent.tool
async def hang_up(ctx: RunContext[None]) -> None:
assert ctx.realtime_session is not None
await ctx.realtime_session.close()
세션은 깨끗하게 닫히고, session.result와 그 이력은 컨텍스트가 종료되기 전에 확정돼요. 툴은 close() 후에 재개되지 않아요. 그것의 결과를 받을 프로바이더가 남아 있지 않으므로 중단된 결과로 로컬에 기록돼요. session() 컨텍스트를 소유한 코드는 예외를 받지 않아요.
대신 실행을 중단하려면 ctx.cancel()이 표준 실행에서처럼 작동해요. 호출이 마찬가지로 중단된 것으로 기록되고, session() 컨텍스트는 완성된 이력을 담은 RunCancelled를 발생시켜요.
호출 중 작업 위임
Realtime 모델은 구조화된 출력을 제공하지 않고, 복잡한 추론에서는 프론티어 텍스트 모델보다 약할 수 있어요. 어려운 작업을 output_type을 가진 표준 Agent에 위임하는 툴을 노출하세요:
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.realtime import RealtimeTurnCompleteEvent
class Answer(BaseModel):
summary: str
confidence: float
supervisor = Agent('openai:gpt-5', output_type=Answer)
voice = Agent(instructions='Answer using the `consult` tool, then read the summary aloud.')
@voice.tool_plain
async def consult(question: str) -> str:
result = await supervisor.run(question)
return result.output.summary
async def main():
async with voice.realtime('openai:gpt-realtime').session() as session:
await session.send(
'Which of our three shipping options is cheapest for a 4 kg parcel to Berlin?'
)
async for event in session:
if isinstance(event, RealtimeTurnCompleteEvent):
break
위임된 실행은 동시에 실행되어, 비동기 툴 호출이 있는 프로바이더는 분석이 실행되는 동안 계속 말할 수 있어요. 음성 세션 후 전체 대화를 계속하려면 이력과 핸드오프를 참고하세요.
엣지 케이스
- 응답은 말한 다음 툴을 호출할 수 있어요. 그것의 음성은 툴 본문이 실행되기 전에 완성되고(출력 전사가 켜져 있으면
stream_transcripts()에 도달해서), 툴 안의 "에이전트가 말했나?" 확인은 이미 그 응답의 음성을 포함해요. - 툴이 끝나는 것이 반드시 턴이 끝나는 것은 아니에요. 턴 경계 참고.
- 짧은 툴은 비동기 Gemini 툴 호출을 역효과로 만들 수 있어요. 결과가 거의 시작하지 않은 응답을 중단할 수 있으니까요. 그렇지 않으면 죽은 공기를 만들 지연이 있는 툴에만 활성화하세요.
- 네이티브 툴 동작은 모델별이에요. 프로바이더의 모든 모델이 같은 툴을 지원한다고 가정하지 말고 프로필과 프로바이더 페이지를 확인하세요.