Human-in-the-loop

Human-in-the-loop (사람 개입 흐름)

사람이 민감한 도구 호출을 승인·거부할 때까지 에이전트 실행을 일시 중지하려면 human-in-the-loop(HITL) 흐름을 쓰세요. 도구가 언제 승인이 필요한지 선언하고, 실행 결과가 대기 중인 승인을 interruption으로 표면화하며, RunState로 일시 중지된 실행을 직렬화하고 결정이 내려진 뒤 재개할 수 있어요.

이 승인 표면은 실행 전체에 걸친 것으로, 현재 최상위 에이전트에 국한되지 않아요. 같은 패턴이 현재 에이전트에 속한 도구, handoff를 통해 도달한 에이전트의 도구, 중첩 Agent.as_tool() 실행의 도구에도 적용돼요. 중첩 Agent.as_tool() 경우 interruption은 여전히 바깥 실행에 표면화되므로, 바깥 RunState에서 승인·거부하고 원래 최상위 실행을 재개하면 돼요.

Agent.as_tool()에서는 승인이 두 계층에서 일어날 수 있어요. 에이전트 도구 자체가 Agent.as_tool(..., needs_approval=...)로 승인을 요구할 수 있고, 중첩 에이전트 안의 도구가 중첩 실행이 시작된 뒤 자기 승인을 일으킬 수도 있어요. 둘 다 같은 바깥 실행 interruption 흐름으로 처리돼요.

이 페이지는 interruption을 통한 수동 승인 흐름에 초점을 맞춰요. 앱이 코드에서 결정할 수 있다면 일부 도구 타입은 프로그램 방식 승인 콜백도 지원해서, 실행이 일시 중지 없이 계속될 수 있어요.

출처: 문서

본문

승인이 필요한 도구 표시

항상 승인을 요구하려면 needs_approval을 True로 설정하거나, 호출마다 결정하는 async 함수를 제공하세요. 이 callable은 실행 컨텍스트, 파싱된 도구 파라미터, 도구 호출 ID를 받아요.

callable 승인 규칙은 SDK가 인자를 안전하게 검사할 수 없을 때 fail-closed로 동작해요. 인자가 없거나, 비어 있거나, 공백만 있거나, 잘못된 JSON이거나, 유효한 JSON인데 객체가 아니거나(null·리스트 같은 것), NaN, Infinity, -Infinity 같은 비표준 상수를 담고 있으면 callable을 호출하지 않고 그 호출은 수동 승인을 요구해요. 이 동작은 Runner와 Realtime 도구 호출에서 같아요.

from agents import Agent
from agents.decorators import tool

@tool(needs_approval=True)
async def cancel_order(order_id: int) -> str:
    return f"Cancelled order {order_id}"

async def requires_review(_ctx, params, _call_id) -> bool:
    return "refund" in params.get("subject", "").lower()

@tool(needs_approval=requires_review)
async def send_email(subject: str, body: str) -> str:
    return f"Sent '{subject}'"

agent = Agent(
    name="Support agent",
    instructions="Handle tickets and ask for approval when needed.",
    tools=[cancel_order, send_email],
)

needs_approval은 function_tool, Agent.as_tool, ShellTool, ApplyPatchTool에서 사용할 수 있어요. 로컬 MCP 서버도 MCPServerStdio, MCPServerSse, MCPServerStreamableHttp의 require_approval로 승인을 지원해요. 호스팅 MCP 서버는 HostedMCPTool에 tool_config={"require_approval": "always"}와 선택적 on_approval_request 콜백으로 승인을 지원해요. Shell·apply_patch 도구는 interruption을 표면화하지 않고 자동 승인·자동 거부하려면 on_approval 콜백을 받아요.

승인 흐름이 동작하는 방식

모델이 도구 호출을 내보내면 러너는 그 승인 규칙(needs_approval, require_approval, 또는 호스팅 MCP 동등물)을 평가해요.
그 도구 호출에 대한 승인 결정이 이미 RunContextWrapper에 저장돼 있다면 러너는 묻지 않고 진행해요. 호출별 승인은 특정 호출 ID에 범위가 지정돼요. 실행의 나머지 동안 같은 도구 identity에 대한 향후 호출에도 같은 결정을 유지하려면 always_approve=True나 always_reject=True를 넘기세요.
승인 규칙이 승인을 요구하고 그 도구 호출에 대한 결정이 저장돼 있지 않으면 실행이 일시 중지되고, RunResult.interruptions(또는 RunResultStreaming.interruptions)에 agent.name, tool_name, arguments 같은 세부 정보를 담은 ToolApprovalItem 항목이 생겨요. 여기에는 handoff 뒤에 또는 중첩 Agent.as_tool() 실행 안에서 일어난 승인도 포함돼요.
결과를 result.to_state()로 RunState로 바꾸고 state.approve(...)나 state.reject(...)를 호출한 뒤, 실행의 원래 최상위 에이전트인 agent로 Runner.run(agent, state) 또는 Runner.run_streamed(agent, state)로 재개해요.
재개된 실행은 멈춘 곳에서 계속되고, 새 승인이 필요하면 이 흐름을 다시 진입해요.

always_approve=True나 always_reject=True로 만든 고정(sticky) 결정은 실행 상태에 저장되므로, 나중에 같은 일시 중지 실행을 재개할 때 state.to_string() / RunState.from_string(...)와 state.to_json() / RunState.from_json(...)을 견뎌요.

HostedMCPTool의 승인 요청에서 Agents SDK는 server_label과 도구 이름의 조합으로 고정 도구 결정을 식별해요. 한 호스팅 MCP 서버에서 lookup_account에 대한 always-approve 결정이 다른 서버의 같은 이름의 도구를 승인하지 않아요. Agents SDK는 호스팅 MCP 승인 요청이 두 비어있지 않은 identity 필드를 모두 포함할 때만 always-approve·always-reject 결정을 영속화해요.

같은 패스에서 모든 대기 승인을 해결할 필요는 없어요. interruptions는 일반 function 도구·호스팅 MCP 승인·중첩 Agent.as_tool() 승인이 섞여 있을 수 있어요. 일부 항목만 승인·거부한 뒤 다시 실행하면, 해결된 호출은 계속되고 해결되지 않은 호출은 interruptions에 남아 실행을 다시 일시 중지해요.

커스텀 거부 메시지

기본적으로 거부된 도구 호출은 SDK의 표준 거부 텍스트를 실행 안으로 되돌려요. 이 메시지는 두 계층에서 커스터마이즈할 수 있어요.

실행 전체 폴백: RunConfig.tool_error_formatter을 설정해 실행 전체에 걸친 승인 거부의 기본 모델 표시 메시지를 제어해요.
호출별 재정의: 특정 거부된 도구 호출에 다른 메시지를 표시하려면 state.reject(...)에 rejection_message=...를 넘기세요.

둘 다 제공되면 호출별 rejection_message가 실행 전체 포매터보다 우선해요.

from agents import RunConfig, ToolErrorFormatterArgs

def format_rejection(args: ToolErrorFormatterArgs[None]) -> str | None:
    if args.kind != "approval_rejected":
        return None
    return "Publish action was canceled because approval was rejected."

run_config = RunConfig(tool_error_formatter=format_rejection)

# Later, while resolving a specific interruption:
state.reject(
    interruption,
    rejection_message="Publish action was canceled because the reviewer denied approval.",
)

두 계층을 함께 보여주는 완전한 예제는 examples/agent_patterns/human_in_the_loop_custom_rejection.py를 참고하세요.

자동 승인 결정

수동 interruption이 가장 일반적인 패턴이지만 유일한 패턴은 아니에요.

로컬 ShellTool과 ApplyPatchTool은 on_approval로 코드에서 즉시 승인·거부할 수 있어요.
HostedMCPTool은 같은 종류의 프로그램 방식 결정을 위해 tool_config={"require_approval": "always"}와 on_approval_request를 함께 쓸 수 있어요.
일반 function_tool 도구와 Agent.as_tool()은 이 페이지의 수동 interruption 흐름을 사용해요.

이 콜백들이 결정을 반환하면 실행은 사람의 응답을 기다리지 않고 계속돼요. Realtime·voice 세션 API의 승인 흐름은 Realtime 가이드를 참고하세요.

스트리밍과 세션

같은 interruption 흐름이 스트리밍 실행에서도 동작해요. 스트리밍 실행이 일시 중지된 뒤에는 반복자가 끝날 때까지 RunResultStreaming.stream_events()를 계속 소비하고, RunResultStreaming.interruptions를 검사해 해결한 다음, 재개 출력이 계속 스트리밍되길 원하면 Runner.run_streamed(...)로 재개하세요. 이 패턴의 스트리밍 버전은 Streaming을 참고하세요.

세션도 쓰고 있다면 RunState에서 재개할 때 같은 세션 인스턴스를 계속 넘기거나, 같은 세션 ID와 백킹 저장소로 구성된 다른 세션 객체를 넘기세요. 그러면 재개된 턴이 같은 저장된 대화 기록에 추가돼요. 세션 lifecycle 세부 사항은 Sessions를 참고하세요.

예제: 일시 중지, 승인, 재개

아래 스니펫은 JavaScript HITL 가이드를 반영해요. 도구가 승인이 필요할 때 일시 중지하고, 상태를 디스크에 영속화하고, 다시 로드한 뒤 결정을 모아 재개해요.

import asyncio
import json
from pathlib import Path

from agents import Agent, Runner, RunState
from agents.decorators import tool

async def needs_oakland_approval(_ctx, params, _call_id) -> bool:
    return "Oakland" in params.get("city", "")

@tool(needs_approval=needs_oakland_approval)
async def get_temperature(city: str) -> str:
    return f"The temperature in {city} is 20° Celsius"

agent = Agent(
    name="Weather assistant",
    instructions="Answer weather questions with the provided tools.",
    tools=[get_temperature],
)

STATE_PATH = Path(".cache/hitl_state.json")

def prompt_approval(tool_name: str, arguments: str | None) -> bool:
    answer = input(f"Approve {tool_name} with {arguments}? [y/N]: ").strip().lower()
    return answer in {"y", "yes"}

async def main() -> None:
    result = await Runner.run(agent, "What is the temperature in Oakland?")

    while result.interruptions:
        # Persist the paused state.
        state = result.to_state()
        STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        STATE_PATH.write_text(state.to_string())

        # Load the state later (could be a different process).
        stored = json.loads(STATE_PATH.read_text())
        state = await RunState.from_json(agent, stored)

        for interruption in result.interruptions:
            approved = await asyncio.get_running_loop().run_in_executor(
                None, prompt_approval, interruption.name or "unknown_tool", interruption.arguments
            )
            if approved:
                state.approve(interruption, always_approve=False)
            else:
                state.reject(interruption)

        result = await Runner.run(agent, state)

    print(result.final_output)

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

이 예제에서 prompt_approval은 input()을 쓰기 때문에 동기이고 run_in_executor(...)로 실행돼요. 승인 소스가 이미 비동기라면(예: HTTP 요청이나 async DB 쿼리) async def 함수를 쓰고 직접 await하면 돼요.

승인으로 일시 중지될 수 있는 실행에서 스트리밍을 쓰려면 Runner.run_streamed를 호출하고 result.stream_events()가 완료될 때까지 소비한 뒤, 위에 보인 것과 같은 result.to_state()와 재개 단계를 따르세요.

저장소 패턴과 예제

스트리밍 승인: examples/agent_patterns/human_in_the_loop_stream.py은 stream_events()를 배출한 뒤 재개 전에 대기 도구 호출을 승인하는 법을 보여줘요. Runner.run_streamed(agent, state)로 재개해요.
커스텀 거부 텍스트: examples/agent_patterns/human_in_the_loop_custom_rejection.py은 승인이 거부될 때 실행 수준 tool_error_formatter과 호출별 rejection_message 재정의를 결합하는 법을 보여줘요.
에이전트-as-도구 승인: Agent.as_tool(..., needs_approval=...)은 위임된 에이전트 작업이 검토를 필요로 할 때 같은 interruption 흐름을 적용해요. 중첩 interruption은 여전히 바깥 실행에 표면화되므로, 중첩 에이전트가 아니라 원래 최상위 에이전트를 재개하세요.
로컬 shell·apply_patch 도구: ShellTool과 ApplyPatchTool도 needs_approval을 지원해요. state.approve(interruption, always_approve=True)나 state.reject(..., always_reject=True)로 실행의 나머지 동안 그 도구에 대한 향후 호출의 결정을 캐시할 수 있어요. 콜백 안에서 승인을 해결하려면 on_approval을 제공하세요. examples/tools/shell.py는 기본적으로 운영자에게 묻는 대화형 콜백을 보여줘요. 모든 shell 호출을 거부하는 자동 정책을 쓰려면 ShellTool에 needs_approval=True와 on_approval=lambda _context, _item: {"approve": False, "reason": "Disabled by policy"}를 설정하세요. 앱이 일시 중지된 실행을 검토하게 하려면 interruption을 처리하세요(examples/tools/shell_human_in_the_loop.py 참고). 호스팅 shell 환경은 needs_approval이나 on_approval을 지원하지 않아요. tools 가이드를 참고하세요.
로컬 MCP 서버: MCPServerStdio / MCPServerSse / MCPServerStreamableHttp에 require_approval을 사용해 MCP 도구 호출을 게이트하세요(examples/mcp/get_all_mcp_tools_example/main.py, examples/mcp/tool_filter_example/main.py 참고).
호스팅 MCP 서버: HostedMCPTool에 tool_config={"require_approval": "always"}를 설정해 HITL을 강제하고, 선택적으로 on_approval_request로 자동 승인·거부하세요(examples/hosted_mcp/human_in_the_loop.py, examples/hosted_mcp/on_approval.py 참고). 신뢰하는 서버에는 "never"를 쓰세요(examples/hosted_mcp/simple.py).
세션과 메모리: Runner.run에 세션을 넘겨 승인과 대화 기록이 여러 턴을 견디게 하세요. SQLite와 OpenAI Conversations 세션 변형은 examples/memory/memory_session_hitl_example.py와 examples/memory/openai_session_hitl_example.py에 있어요.
Realtime 에이전트: realtime 데모는 RealtimeSession의 approve_tool_call / reject_tool_call로 도구 호출을 승인·거부하는 WebSocket 메시지를 노출해요. 서버 측 핸들러는 examples/realtime/app/server.py, API 표면은 Realtime 가이드를 참고하세요.

장기 실행 승인 (Long-running approvals)

RunState는 지속(durable) 가능하도록 설계됐어요. state.to_json()이나 state.to_string()으로 대기 중인 작업을 데이터베이스·큐에 저장하고, 나중에 RunState.from_json(...)이나 RunState.from_string(...)으로 다시 만들어요.

승인 상태를 서버에 두기

직렬화된 RunState는 승인 결정·대기 중인 도구 호출·도구 인자를 포함한 실행 상태를 담아요. SDK는 이 상태를 복원하는데, RunState.from_json()과 RunState.from_string()은 스냅샷이나 제출하는 사람을 인증하지 않아요. 신뢰하는 저장소의 스냅샷이나, 애플리케이션이 완전한 무결성·소유권을 검증한 스냅샷만 역직렬화하세요. 스키마 검사나 도구 호출 지문(fingerprint)은 스냅샷을 인증하지 않아요.

브라우저·모바일 승인 인터페이스에서는 완전한 스냅샷을 애플리케이션 제어 서버 저장소에 두세요. 검토자에게는 검토자가 볼 권한이 있는 도구 세부 정보와 대기 결정의 불투명 ID만 보내세요. 도구 이름과 인자는 신뢰할 수 없는 표시 콘텐츠로 취급하고 HTML 렌더링 시 이스케이프하세요.

결정이 도착하면 서버는 반드시:

애플리케이션의 세션·인증 미들웨어로 검토자를 인증하세요. 승인 요청 본문에서 검토자 신원을 가져오지 마세요.
그 검토자가 저장된 실행과 선택된 대기 호출에 대한 권한이 있는지 확인하세요. 실행 ID나 결정 ID를 가지고 있는 것은 인증이 아니에요.
제출된 결정 식별자와 boolean 결정을 서버에 저장된 대기 요청에 대해 검증하세요. 서버가 소유한 스냅샷을 로드하고 state.get_interruptions()로 대기 항목을 얻으세요. 클라이언트의 교체 도구 호출·인자·승인 기록·직렬화 상태를 받아들이지 마세요.
서버가 소유한 항목에 state.approve(...)나 state.reject(...)를 적용한 뒤 실행을 재개하세요. 동시·재생 제출이 같은 스냅샷을 두 번 재개하지 않도록 각 대기 요청의 소비를 저장소와 조정하세요. 공유 저장소에서는 재개 실행을 시작하기 전에 원자적 owner-checked 전환을 쓰세요.

서버 측 승인 예제는 CLI 클라이언트 시뮬레이션과 한 프로세스·한 이벤트 루프에 국한된 store로 이 패턴을 보여줘요. 배치의 모든 대기 호출에 대해 결정 하나를 요구해요. 예제는 역직렬화와 재개 실행 전에 요청을 소비하므로, 실패와 취소도 요청을 소비해요. 프로덕션 애플리케이션은 재시도 전에 도구 부수 효과를 조정하는 인증·요청 보호·저장소 보존·복구를 제공해야 해요. 이 예제는 배포 가능한 HTTP 서비스가 아니에요.

context를 context_override로 교체하거나 strict_context=True를 설정하거나 직렬화된 승인 기록만 제거하는 것은 신뢰할 수 없는 스냅샷을 안전하게 만들지 않아요. 다른 필드가 여전히 재개 실행을 제어해요. 애플리케이션이 완전한 스냅샷을 클라이언트로 운반한다면, 역직렬화 전에 무결성을 검증하고 승인된 사용자·실행에 바인딩하며 재생을 막아야 해요. 그런 검증은 스냅샷을 암호화하거나 그 내용을 클라이언트로부터 숨기지 않아요.

직렬화 옵션

유용한 직렬화 옵션:

context_serializer — 매핑이 아닌 컨텍스트 객체가 직렬화되는 방식을 커스터마이즈.
context_deserializer — RunState.from_json(...)이나 RunState.from_string(...)으로 상태를 로드할 때 매핑이 아닌 컨텍스트 객체를 재구성.
strict_context=True — 컨텍스트가 이미 매핑이거나 context_serializer를 제공하지 않으면 직렬화 실패. 컨텍스트가 이미 매핑이거나 context_deserializer를 제공하지 않으면 역직렬화 실패.
context_override — 상태를 로드할 때 직렬화된 컨텍스트를 교체. 원래 컨텍스트 객체를 복원하고 싶지 않을 때 유용하지만, 이미 직렬화된 페이로드에서 그 컨텍스트를 제거하지는 않아요.
include_tracing_api_key=True — 재개된 작업이 같은 자격 증명으로 trace를 계속 내보내야 할 때 직렬화된 trace 페이로드에 tracing API 키 포함.

직렬화된 실행 상태는 앱 컨텍스트에 더해 승인·사용량·직렬화된 tool_input·중첩 에이전트-as-도구 재개·trace 메타데이터·서버 관리 대화 설정 같은 SDK 관리 런타임 메타데이터를 포함해요. 직렬화된 상태를 저장·전송할 계획이라면 RunContextWrapper.context를 영속 데이터로 취급하고, 의도적으로 상태와 함께 보내고 싶지 않다면 거기에 비밀을 두지 마세요.

대기 중인 작업 버전 관리

승인이 한동안 기다릴 수 있다면 직렬화된 상태 옆에 에이전트 정의나 SDK용 버전 마커를 저장하세요. 그러면 모델·프롬프트·도구 정의가 바뀌어도 불일치를 피하도록 역직렬화를 일치하는 코드 경로로 라우팅할 수 있어요.

더 알아보기 (Learn more)

OpenAI Agents SDK 문서에서 더 많은 가이드를 확인하세요.