커스텀 캐퍼빌리티 만들기

커스텀 캐퍼빌리티 만들기 (Building Custom Capabilities)

자신만의 캐퍼빌리티를 만들려면 AbstractCapability를 서브클래싱하고 필요한 메서드를 오버라이드하세요. 이 문서에서는 캐퍼빌리티의 설정 메서드와 라이프사이클 훅을 하나씩 살펴보며, 직접 만들어 조립하는 방법을 알려 드릴게요.

출처: 문서

본문

자신만의 캐퍼빌리티를 만들려면 AbstractCapability를 서브클래싱하고 필요한 메서드를 오버라이드하세요. 두 가지 범주가 있어요: 에이전트 생성 시 호출되는 설정 메서드 — 그리고 for_run이 교체 인스턴스를 반환하면 런 설정 시 그 위에서 다시 실행됩니다 (참고: 런별 상태 격리); get_wrapper_toolset은 항상 런별로 호출됩니다 — 그리고 각 런 중에 발동하는 라이프사이클 훅.

커스텀 캐퍼빌리티 클래스는 일반 클래스 또는 dataclass일 수 있어요. 공유 메타데이터 속성 — id, description, defer_loading — 은 항상 켜진 캐퍼빌리티의 캐퍼빌리티 객체에 대한 선택적 선언입니다. id가 생략되면 Pydantic AI가 클래스 이름에서 런 로컬 id를 유도하고 런 내에서 중복을 구분합니다. 지연 캐퍼빌리티는 명시적이고 안정적인 id가 필요해요.

단일 고정 관심사를 다루는 캐퍼빌리티는 대신 안정적인 기본 id를 선언합니다 — 내장 WebSearch는 'web_search', Thinking은 'thinking'을 쓰는 식 — 그래서 지속 실행이 여러분이 만들지 않은 것을 이름 짓지 않고도 그것들이 기여하는 바를 식별할 수 있어요. 자신의 캐퍼빌리티도 같은 방식으로 일회성일 때 기본 id를 주세요.

id가 고정되어 있어서, 두 개가 하나의 id 아래에서 만날 수 있고 그것이 무엇을 뜻하는지는 그것들이 어디서 왔는지에 달려 있습니다.

같은 에이전트의 두 개는 하나의 설정을 두 번 말한 것이므로, 필드별로 병합됩니다. 하나만 말한 값은 유지되고, 둘 다 말한 값은 나중 것을 취해요. 그래서 두 개의 패키지된 캐퍼빌리티가 각각 WebSearch를 가져오고 에이전트가 두 도메인 집합에 모두 닿을 수 있는 것입니다. 기본 id 자체 외에 선언할 것이 없어요 — 하나를 쓰는 것 자체가 에이전트에 그것이 하나 있다는 진술입니다:

from dataclasses import dataclass
from typing import Any

from pydantic_ai.capabilities import AbstractCapability


@dataclass
class Retries(AbstractCapability[Any]):
    limit: int = 3
    id: str | None = 'retries'

두 인스턴스가 하나의 설정에 대한 두 진술이 아니라 두 정체성 일 때 — 두 계정, 두 자격 증명 — 고정된 기본 id는 잘못된 형태입니다. 병합하면 하나가 조용히 사라지기 때문이에요. 그것을 구분하는 것으로 id를 유도하세요. 그러면 두 정체성은 두 id를 지니고 두 캐퍼빌리티로 남고, 하나의 id 아래 두 개는 진짜 실수라 보고됩니다. MCP는 서버의 호스트와 마지막 경로 세그먼트에서 유도하므로, 포트나 이른 경로 세그먼트에서만 다른 서버는 여전히 구별되는 명시적 id가 필요해요.

구성이 필드 병합 이상일 때만 combine을 오버라이드하세요 — 두 값의 더 작은 것을 취해야 하는 예산 같은 경우요. 기본 병합은 dataclass 필드만 봅니다. 그래서 고정 기본 id와 인스턴스 설정이 있는 일반 클래스는 combine을 오버라이드해야 해요. 그렇지 않으면 반복 인스턴스가 그 설정을 조용히 버리는 대신 오류를 냅니다. 앞의 언더스코어는 그것을 바꾸지 않아요. 중요한 것은 병합이 속성을 열거할 수 있는지이지, 그것이 private인지가 아니에요.

그 필드들에서 유도하는 상태는 예외이며, cached_property가 그렇게 말하는 방법입니다. 병합은 캐시된 값을 버리므로, 다음 읽기는 병합된 필드에 대해 재계산합니다 — 그것은 __post_init__이 줄 수 없는 답입니다. 병합이 의도적으로 그것을 다시 실행하지 않기 때문이죠. 필드여야 하는 상태는 자신의 combine에서 재계산하세요.

런을 위해 공급된 캐퍼빌리티는 에이전트 수준의 동명을 완전히 덮어씁니다 — agent.run(capabilities=[Thinking(effort='high')])는 에이전트의 Thinking과 병합하는 대신 그것을 대체합니다. 런은 이 런이 무엇을 하는지 말하므로, 병합하면 런이 대체하려던 에이전트 수준 설정이 살아남고, 런이 부과하려던 제한을 에이전트 수준 허용 목록이 넓힐 수 있어요. combine은 계층을 가로질러 조회되지 않습니다. 무관한 캐퍼빌리티 클래스는 계층 간에 id를 공유할 수 없어요. 캡슐화된 캐퍼빌리티의 id를 유지하는 투명 래퍼(prefix_tools() 같은)는 계층 간에 그 캐퍼빌리티를 대체할 수 있지만, 이것이 서로 다른 래퍼 클래스가 한 계층 안에서 병합되게 하지는 않습니다.

두 개를 병합 대신 유지하려면 각각에 구별되는 id를 전달하거나, 유도·구분된 id로 되돌아가려면 id=None을 전달하세요. 기본 id를 선언하지 않는 캐퍼빌리티에 전달하는 id는 여러분이 고른 이름이므로, 같은 것을 두 번 전달하면 병합이 아니라 충돌로 보고됩니다.

한 가지 한계: 네이티브 도구를 기여하는 캐퍼빌리티의 경우, 이 탈출구들이 네이티브 도구 자신의 id를 풀어주지 않아요. 다르게 구성된 WebSearch 중복 두 개는 캐퍼빌리티에 구별되는 id를 주든 id=None을 주든 네이티브 도구 id 충돌을 여전히 냅니다 — 동일한 구성만이 그것을 피해요.

from typing import Any

from pydantic_ai.capabilities import AbstractCapability


class MyCapability(AbstractCapability[Any]):
    """A custom capability."""

자체 설정 필드나 공유 메타데이터 필드에 생성된 생성자 파라미터를 원할 때는 dataclass를 쓰세요:

from dataclasses import dataclass

from pydantic_ai.capabilities import AbstractCapability


@dataclass
class MyCapability(AbstractCapability[None]):
    label: str

커스텀 __init__을 정의하면 노출하려는 메타데이터만 설정하세요. super().__init__()나 __post_init__() 요구사항은 없어요:

from pydantic_ai.capabilities import AbstractCapability


class MyCapability(AbstractCapability[None]):
    def __init__(
        self,
        label: str,
        *,
        id: str | None = None,
        description: str | None = None,
        defer_loading: bool = False,
    ) -> None:
        self.id = id
        self.description = description
        self.defer_loading = defer_loading
        self.label = label

defer_loading=True일 때 안정적이고 명시적인 id를 제공하세요. 히스토리 재생이 그것에 의존하며, Pydantic AI는 없는 지연 캐퍼빌리티를 거부합니다. 항상 켜진 캐퍼빌리티에서는 id를 생략해도 클래스 이름에서 런 로컬 id를 유도합니다.

의존성 타이핑 (Typing dependencies)

AbstractCapability은 에이전트의 의존성 타입에 제네릭입니다 — AbstractCapability[MyDeps]는 그 캐퍼빌리티의 훅이 RunContext[MyDeps]를 받음을 의미해요. 캐퍼빌리티가 어떤 의존성 타입과도 동작하면 AbstractCapability[Any]를, 의존성 필드에 접근해야 하면 특정 타입을 쓰세요:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.models.test import TestModel


@dataclass
class UserGreeter(AbstractCapability[Any]):
    """Works with any deps type."""

    async def before_model_request(
        self, ctx: RunContext[Any], request_context: ModelRequestContext
    ) -> ModelRequestContext:
        return request_context


@dataclass
class Deps:
    user_name: str


@dataclass
class PersonalGreeter(AbstractCapability[Deps]):
    """Requires Deps with a user_name field."""

    async def before_model_request(
        self, ctx: RunContext[Deps], request_context: ModelRequestContext
    ) -> ModelRequestContext:
        # ctx.deps is typed as Deps -- IDE autocomplete works
        print(f'Request for {ctx.deps.user_name}')
        #> Request for Alice
        return request_context


agent = Agent(
    TestModel(),
    deps_type=Deps,
    capabilities=[UserGreeter(), PersonalGreeter()],
)
agent.run_sync('hi', deps=Deps(user_name='Alice'))

도구 제공 (Providing tools)

도구를 제공하는 캐퍼빌리티는 get_toolset에서 툴셋을 반환합니다. 이것은 미리 만들어진 AbstractToolset 인스턴스일 수도, RunContext를 받아 동적으로 하나를 반환하는 callable일 수도 있어요:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.toolsets import AgentToolset, FunctionToolset

math_toolset = FunctionToolset()


@math_toolset.tool_plain
def add(a: float, b: float) -> float:
    """Add two numbers."""
    return a + b


@math_toolset.tool_plain
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b


@dataclass
class MathTools(AbstractCapability[Any]):
    """Provides basic math operations."""

    def get_toolset(self) -> AgentToolset[Any] | None:
        return math_toolset


agent = Agent('openai:gpt-5.2', capabilities=[MathTools()])
result = agent.run_sync('What is 2 + 3?')
print(result.output)
#> The answer is 5.0

네이티브 도구의 경우, get_native_tools를 오버라이드해 AgentNativeTool 인스턴스들(AbstractNativeTool 객체와 RunContext를 받는 callable 모두 포함)의 시퀀스를 반환하세요.

툴셋 감싸기 (Toolset wrapping)

get_wrapper_toolset은 에이전트의 전체 조립된 툴셋을 WrapperToolset으로 감싸게 해 줘요. 이것은 도구를 제공하는 것보다 강력합니다 — 도구 실행을 가로채거나, 로깅을 추가하거나, 횡단 관심사를 적용할 수 있어요.

래퍼는 ( prepare_tools 훅이 감싼 후의) 결합된 비-출력 툴셋을 받습니다. 출력 도구는 별도로 추가되며 영향받지 않아요.

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.toolsets import AbstractToolset
from pydantic_ai.toolsets.wrapper import WrapperToolset


@dataclass
class LoggingToolset(WrapperToolset[Any]):
    """Logs all tool calls."""

    async def call_tool(
        self, tool_name: str, tool_args: dict[str, Any], *args: Any, **kwargs: Any
    ) -> Any:
        print(f'  Calling tool: {tool_name}')
        return await super().call_tool(tool_name, tool_args, *args, **kwargs)


@dataclass
class LogToolCalls(AbstractCapability[Any]):
    """Wraps the agent's toolset to log all tool calls."""

    def get_wrapper_toolset(self, toolset: AbstractToolset[Any]) -> AbstractToolset[Any]:
        return LoggingToolset(wrapped=toolset)


agent = Agent('openai:gpt-5.2', capabilities=[LogToolCalls()])


@agent.tool_plain
def greet(name: str) -> str:
    """Greet someone."""
    return f'Hello, {name}!'


result = agent.run_sync('hello')
# Tool calls are logged as they happen

참고

get_wrapper_toolset은 비-출력 툴셋 을 런당 한 번(툴셋 조립 중) 감쌉니다. prepare_tools와 prepare_output_tools 훅도 PreparedToolset 래퍼를 통해 흐르므로, 세 가지 모두 툴셋 수준에서 통합됩니다 — get_wrapper_toolset은 prepare_tools 주변에서 돌아갑니다(준비된 defs를 봄), 그리고 prepare_output_tools는 출력 툴셋을 독립적으로 감쌉니다.

지침 제공 (Providing instructions)

get_instructions은 에이전트에 지침을 추가합니다. 에이전트 생성 시 한 번 호출되므로, 동적 값이 필요하면 callable을 반환하세요:

from dataclasses import dataclass
from datetime import datetime
from typing import Any

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class KnowsCurrentTime(AbstractCapability[Any]):
    """Tells the agent what time it is."""

    def get_instructions(self):
        def _get_time(ctx: RunContext[Any]) -> str:
            return f'The current date and time is {datetime.now().isoformat()}.'

        return _get_time


agent = Agent('openai:gpt-5.2', capabilities=[KnowsCurrentTime()])
result = agent.run_sync('What time is it?')
print(result.output)
#> The current time is 3:45 PM.

지침은 에이전트의 의존성에 대해 렌더링되는 Handlebars 스타일 템플릿용 템플릿 문자열 (TemplateStr('Hello {{name}}'))도 쓸 수 있어요. 파이썬 코드에서는 IDE 자동완성을 위해 RunContext를 가진 callable이 일반적으로 선호됩니다.

id가 있는 캐퍼빌리티의 지침은 'capability:<capability id>'로 식별되는 자체 instruction parts로 모델에 도달합니다 — 리터럴·계산된 것 모두, 그래서 그 키를 오버라이드하는 애플리케이션은 캐퍼빌리티가 모델에 말하는 모든 것을 제어합니다. 한 파트를 자체로 주소 지정 가능하게 하려면 @capability.instructions(name=...)로 이름을 지어 'capability:<capability id>:<name>'로 키함하세요. id가 없는 캐퍼빌리티는 둘 다에 키가 없어요.

모델 설정 제공 (Providing model settings)

get_model_settings은 모델 설정을 dict로, 또는 스텝별 설정용 callable로 반환합니다.

모델 설정이 스텝별로 달라야 할 때 — 예를 들어 재시도할 때만 씽킹을 켜거나, 도구가 호출될 때까지 특정 tool_choice를 강제할 때 — callable을 반환하세요:

from dataclasses import dataclass

from pydantic_ai import Agent, ModelSettings, RunContext
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class ThinkingOnRetry(AbstractCapability):
    """Enables thinking mode when the agent is retrying."""

    def get_model_settings(self):
        def resolve(ctx: RunContext) -> ModelSettings:
            if ctx.run_step > 1:
                return ModelSettings(thinking='high')
            return ModelSettings()

        return resolve


agent = Agent('openai:gpt-5.2', capabilities=[ThinkingOnRetry()])
result = agent.run_sync('hello')
print(result.output)
#> Hello! How can I help you today?

callable은 ctx.model_settings에 이 캐퍼빌리티 앞에서 해석된 모든 계층의 병합 결과(모델 기본값과 에이전트 수준 설정)가 담긴 RunContext를 받습니다.

모델 선택 (Selecting the model)

모델 선택이 더 큰 커스텀 캐퍼빌리티의 한 부분일 때 get_model()을 오버라이드하세요. Model, 모델 ID 문자열, 또는 ModelSelectionContext를 받는 sync/async callable을 반환합니다. 이 예는 매 요청 스텝마다 의존성에서 모델을 고릅니다:

from __future__ import annotations

from dataclasses import dataclass
from typing import Literal

from pydantic_ai import Agent, ModelSelectionContext
from pydantic_ai.capabilities import AbstractCapability, ModelSelector


@dataclass
class Deps:
    """Dependencies that influence model selection."""

    task_complexity: Literal['standard', 'complex']


class AdaptiveModel(AbstractCapability[Deps]):
    """Select a model for each request step."""

    def get_model(self) -> ModelSelector[Deps]:
        return self.select_model

    def select_model(self, ctx: ModelSelectionContext[Deps]) -> str:
        return 'openai:gpt-5.6-sol' if ctx.deps.task_complexity == 'complex' else 'openai:gpt-5.6-luna'


agent = Agent(deps_type=Deps, capabilities=[AdaptiveModel()])

get_model()은 동기 설정 메서드이지만, 반환하는 ModelSelector는 동기 또는 비동기일 수 있어요. ModelSelectionContext는 완전한 런 컨텍스트가 현재 선택 중인 모델을 필요로 하기 때문에 RunContext와 별개입니다. 그것은 의존성, 요청 스텝, 메시지 히스토리, 사용량을 포함합니다. get_model() 자체는 가볍게 유지하세요. I/O는 async 셀렉터에서 수행하세요.

get_model()에서 직접 반환된 모델이나 모델 ID는 런당 한 번 해석됩니다. get_model()에서 반환된 셀렉터는 매 논리 모델 요청 스텝 전에 평가됩니다.

캐퍼빌리티의 모델은 호출 지점 run(model=...) 인수와 런 수준 spec= 모델 아래, 에이전트 생성자의 모델 위에 끼워집니다. 높은 것부터 낮은 것까지:

run()/iter() 인수 › 런 spec= 모델 › 캐퍼빌리티 get_model() › 에이전트 생성자.

override(model=...)는 여전히 이 모두보다 이깁니다. 명시적 모델은 캐퍼빌리티 선택을 완전히 건너뜁니다.

나중의 모델 기여가 이른 것들을 덮어씁니다. for_run()이 캐퍼빌리티를 변경 없이 두면, 그 부트스트랩 선택이 1단계에 재사용되고, 다른 셀렉터가 있는 교체를 반환하면 그 셀렉터가 새로운 1단계 선택을 만듭니다.

폴백은 선택과 상호 보완적입니다. 요청 실패가 다른 모델에서 재시도되어야 하면 설정된 FallbackModel을 반환하세요.

모델 ID 해석 (Resolving model IDs)

애플리케이션 특정 문자열이 커스텀 공급자 구성, 자격 증명, 레지스트리 조회를 필요로 할 때 resolve_model_id()을 오버라이드하세요. 모델 선택과 달리 해석은 first-wins예요: 캐퍼빌리티를 순서대로 시도하고, 모든 해석기가 None을 반환할 때만 일반 infer_model() 동작이 사용됩니다.

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, ModelResolutionContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.models import KnownModelName, Model, infer_model
from pydantic_ai.providers import Provider, infer_provider
from pydantic_ai.providers.openai import OpenAIProvider


@dataclass
class Deps:
    """Per-user provider credentials."""

    openai_api_key: str


class UserModelResolver(AbstractCapability[Deps]):
    """Resolve user-scoped model IDs with per-user credentials."""

    async def resolve_model_id(
        self,
        ctx: ModelResolutionContext[Deps],
        *,
        model_id: KnownModelName | str,
    ) -> Model | None:
        if not model_id.startswith('user:'):
            return None

        def provider_factory(provider_name: str) -> Provider[Any]:
            if provider_name == 'openai':
                return OpenAIProvider(api_key=ctx.deps.openai_api_key)
            return infer_provider(provider_name)

        return infer_model(model_id.removeprefix('user:'), provider_factory)


agent = Agent('user:openai:gpt-5.6-sol', deps_type=Deps, capabilities=[UserModelResolver()])

생성자 ID는 for_agent()를 통해 문자열로 유지되므로, 바인딩된 캐퍼빌리티가 기본 추론이 먼저 다른 구성·자격 증명으로 공급자를 만들지 않고 해석기를 설치할 수 있어요.

해석 결과는 한 런 동안 모델 ID와 해석기 트리별로 캐시됩니다. 스텝별 셀렉터가 같은 문자열을 또 반환하면, Pydantic AI는 해석기를 다시 호출하지 않고 같은 모델·공급자·클라이언트를 재사용합니다. 나중 스텝에서 의도적으로 다르게 해석하려면 다른 ID를 선택하거나 셀렉터에서 Model 인스턴스를 직접 반환하세요.

모델 선택 라이프사이클과 제한

부트스트랩 해석은 for_agent() 바인딩 후, for_run() 전의 캐퍼빌리티 트리를 사용합니다. 첫 모델을 해석하는 것이 완전한 RunContext를 가능하게 하기 때문이에요. for_run()이 교체 캐퍼빌리티를 반환하면, 1단계나 이후 단계에서 선택된 문자열은 그 교체의 해석기 체인을 사용합니다. for_run()이 캐퍼빌리티를 변경 없이 두면, 이미 해석된 부트스트랩 모델이 1단계에 재사용됩니다.

모델 선택과 해석은 eager 훅이므로, 지연 캐퍼빌리티는 로드된 후에도 그것을 기여하지 않아요. 런 스펙 캐퍼빌리티는 부트스트랩 동안 알려져 첫 모델을 공급할 수 있어요. CapabilityFunc 또는 모델을 for_run()에서만 도입하는 다른 캐퍼빌리티는 기존 부트스트랩 모델이 필요합니다 — for_run()이 완전한 RunContext를 받기 때문입니다. 그것은 1단계부터 그 모델을 대체할 수 있지만, 모델 없는 에이전트를 부트스트랩할 수는 없어요. 지연 캐퍼빌리티를 로드한 후 새 모델을 선택하는 것이 애플리케이션에 유용하다면, 원하는 스텝·연속 시맨틱을 설명하는 이슈를 열어 주세요.

동적 선택은 현재 지속 실행 캐퍼빌리티가 지원하지 않습니다. 지속 런은 실행 전에 모델 ID가 등록되어야 하고 재생·크로스 런 재개 중 같은 선택 모델을 재생성해야 해요. 지속 실행에는 명시적 등록 모델을 전달하세요. 별도의 일반 런에서 일시 정지된 공급자 요청을 재개하는 것도, 이전 모델이 셀렉터에서 왔으면 명시적 모델이 필요합니다.

설정 메서드 레퍼런스 (Configuration methods reference)

메서드	반환 타입	목적
`get_toolset()`	`AgentToolset` \\ `None`	도구를 제공할 툴셋
`get_native_tools()`	`Sequence[AgentNativeTool]`	등록할 네이티브 도구(callable 포함)
`get_wrapper_toolset()`	`AbstractToolset` \\ `None`	에이전트의 조립된 툴셋을 감쌀 래퍼 툴셋
`get_instructions()`	`AgentInstructions` \\ `None`	에이전트에 추가할 지침
`get_model_settings()`	`AgentModelSettings` \\ `None`	모델 설정
`get_model()`	`AgentModel` \\ `None`	정적 또는 동적 모델 기여
`resolve_model_id()`	`Model` \\ `None`	애플리케이션 특정 모델 ID 해석

에이전트에 바인딩 (Binding to an agent)

재사용 가능한 캐퍼빌리티가 붙은 에이전트를 검사해야 할 때 for_agent()을 오버라이드하세요. 이 훅은 Agent 생성 중, 에이전트 자신의 모델·이름·툴셋이 사용 가능해진 후 그리고 캐퍼빌리티 기여가 추출되기 전에 한 번 실행됩니다:

from dataclasses import dataclass, replace
from typing import Any

from typing_extensions import Self

from pydantic_ai import Agent
from pydantic_ai.agent import AbstractAgent
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class AgentIdentity(AbstractCapability[Any]):
    """Add the bound agent's name to its instructions."""

    agent_name: str | None = None

    def for_agent(self, agent: AbstractAgent[Any, Any]) -> Self:
        return replace(self, agent_name=agent.name)

    def get_instructions(self) -> str:
        return f'You are the {self.agent_name} agent.'


identity = AgentIdentity()
support = Agent('openai:gpt-5.2', name='support', capabilities=[identity])
sales = Agent('openai:gpt-5.2', name='sales', capabilities=[identity])

for_agent()은 런 의존성이나 라이프사이클 컨텍스트가 존재하기 전, 에이전트 생성 중에 구성을 바인딩하므로 동기입니다. I/O가 없게 유지하세요. 비동기 런별 설정은 for_run()에 속합니다.

같은 캐퍼빌리티가 여러 에이전트에 붙을 수 있다면 원본을 변경하지 말고 새 바인딩 복사본을 반환하세요. CombinedCapability와 WrapperCapability는 바인딩을 자식에 전파하고, 바인딩 복사본은 get_model()과 resolve_model_id()를 포함한 모든 설정 훅에 참여합니다.

파라미터는 AbstractAgent로 타입되어, 재사용 가능한 캐퍼빌리티가 이식 가능한 에이전트 인터페이스에만 의존하고 커스텀 에이전트 구현과 호환되게 합니다. WrapperAgent를 통한 런은 감싸진 에이전트에 위임되므로, Pydantic AI의 내장 래퍼는 캐퍼빌리티를 외부 래퍼에 다시 바인딩하지 않아요.

for_agent()은 생성자 모델을 호출자가 공급한 그대로 정확히 봅니다. 특히 모델 ID는 바인딩이 실행되는 동안 문자열로 남습니다. 그래서 바인딩된 캐퍼빌리티가 기본 추론이 먼저 잘못된 구성·자격 증명으로 공급자를 만들지 않고 resolve_model_id()를 도입할 수 있어요. 바인딩된 캐퍼빌리티 트리에 해석기가 없고 defer_model_check=False면, 바인딩 후 일반 모델 추론이 일어납니다.

run()에 직접 전달되거나 런 스펙을 통해 전달된 캐퍼빌리티는 부트스트랩 모델 선택 전에 그 런에 대해 한 번 바인딩됩니다. CapabilityFunc은 런 전에 자체로 바인딩됩니다. 그 반환 값이 보통 독립적으로 재사용 가능한 캐퍼빌리티이므로, 그 값도 자체 for_run()이 호출되기 전에 바인딩됩니다. 대조적으로, 일반 캐퍼빌리티의 for_run()이 반환하는 특수한 런 바인딩 캐퍼빌리티는 for_agent()을 다시 통과하지 않아요.

캐퍼빌리티 라이프사이클 (Capability lifecycle)

바인딩 훅은 어떤 캐퍼빌리티가 런에 참여하는지 정하고, 라이프사이클 훅은 그다음 그것이 수행하는 작업을 가로챕니다. 높은 수준 순서는:

for_agent() → 부트스트랩 모델 선택·해석 → for_run() → 스텝별 선택·준비 → 모델 요청 → 도구/출력 처리 → 런 완료

단계	캐퍼빌리티 작업	사용 가능한 것
에이전트 바인딩	`for_agent()`	에이전트 이름, 원시 생성자 모델, 툴셋, 기타 생성자 구성; 런 의존성이나 `RunContext` 없음
런 부트스트랩	`get_model()`, 선택이 문자열이면 이어서 `resolve_model_id()`	의존성, 메시지 히스토리, 사용량, 더 낮은 우선순위의 모델을 선택·해석 컨텍스트를 통해; 아직 완전한 `RunContext` 없음
런 바인딩	`for_run()`	부트스트랩 모델을 담은 완전한 `RunContext`; 런 범위 교체 캐퍼빌리티를 반환할 수 있음
각 논리 모델 스텝	`for_run()` 후 모델 선택·해석, 모델 설정, 도구 준비, 메시지 준비	선택된 모델은 그 설정·프로파일 민감 도구·모델별 메시지 준비가 평가되기 전에 `RunContext`에 설치됨
모델 요청과 응답	모델 요청·도구·출력·노드·이벤트 스트림 훅	완전히 준비된 요청과 각 훅에 적합한 실시간 런 상태
런 완료	`after_run`, `on_run_error`, `wrap_run` 완료	최종 결과 또는 오류, 누적 메시지, 사용량

for_run()이 원래 캐퍼빌리티를 반환하면, 부트스트랩 모델 선택이 1단계에 재사용됩니다. 교체 캐퍼빌리티는 1단계에 다른 모델을 선택할 수 있습니다. 한 논리 스텝 안의 연속 폴링은 그 스텝의 선택된 모델에 고정됩니다.

라이프사이클에 훅 걸기 (Hooking into the lifecycle)

캐퍼빌리티는 각각 최대 네 가지 변형이 있는 다섯 개의 라이프사이클 지점에 훅을 걸 수 있어요:

before_* — 동작 전에 발동, 입력을 수정 가능
after_* — 동작 성공 후(캐퍼빌리티 역순)에 발동, 출력을 수정 가능
wrap_* — 완전한 미들웨어 제어: handler callable을 받고 그것을 호출할지·어떻게 호출할지 결정
on_*_error — 동작이 실패할 때(wrap_*이 복구 기회를 가진 후)에 발동, 오류를 관찰·변환·복구 가능

팁

서브클래싱 없는 빠른 애플리케이션 수준 훅은 Hooks 캐퍼빌리티를 쓰세요.

런 훅 (Run hooks)

훅	시그니처	목적
`before_run`	`(ctx:` `RunContext) -> None`	런이 시작한다는 관찰 전용 통지
`after_run`	`(ctx:` `RunContext, *, result:` `AgentRunResult) ->` `AgentRunResult`	최종 결과 수정
`wrap_run`	`(ctx:` `RunContext, *, handler:` `WrapRunHandler) ->` `AgentRunResult`	전체 런 감싸기
`on_run_error`	`(ctx:` `RunContext, *, error: BaseException) ->` `AgentRunResult`	런 오류 처리 (참고: error hooks)

wrap_run은 오류 회복을 지원합니다. handler()가 오류를 일으키고 wrap_run이 그 예외를 잡아 대신 결과를 반환하면, 오류는 억제되고 회복 결과가 사용됩니다. 이것은 agent.run(), agent.iter(), realtime 세션과 동작합니다 — realtime 세션은 런이라 네 훅 모두 그것 주변에서 한 번 발동하고, wrap_run의 핸들러는 세션이 닫힐 때 해결됩니다. ctx.realtime을 확인해 동작을 분기하고, (세션이 연결되면 설정되는) ctx.realtime_session으로 실시간 세션과 상호작용하세요.

여러분이 생성한 태스크 정리

런은 하나의 asyncio 태스크를 취소함으로써 취소됩니다 — RunContext.cancel(), CancellationToken, asyncio.wait_for 타임아웃, 또는 둘러싼 태스크 그룹으로요. 런이 인라인으로 await하는 작업은 CancelledError를 자동으로 받지만, asyncio.create_task(...)로 직접 시작한 태스크는 다른 태스크에서 실행되어 그렇게 받지 않아요. 그래서 태스크를 생성하는 캐퍼빌리티는 스스로 그것들을 정리해야 합니다.

구조적 동시성(anyio.create_task_group()과 async with)을 선호하세요. 런의 취소가 async with를 통해 흐르고 자식들이 범위 종료 시 취소되어 수동 정리가 필요 없어요. 원시 태스크를 유지한다면 wrap_run의 try/finally에서 취소하고 배수하세요 (먼저 모든 task.cancel()을 발행, 그다음 단일 await asyncio.gather(*tasks, return_exceptions=True)), 그리고 그 정리를 anyio.CancelScope(shield=True)로 감싸 런이 이미 취소되는 중이어도 완료되게 하세요. 원시 task.cancel()은 shielded 범위도 뚫을 수 있으므로, 반드시 끝나야 하는 작업은 자체 태스크에 두고 asyncio.shield()로 보호하세요(태스크에 강한 참조 유지) — 태스크를 직접 await하는 것은 도움이 안 됩니다. 런의 태스크를 취소하는 것이 그것이 await하는 태스크로 CancelledError를 전파하기 때문입니다. 백그라운드 태스크에서 시작한 서브에이전트 런도 취소·배수하는 것은 여러분의 몫입니다 — 인라인으로 await하는 서브에이전트만 여러분을 위해 정리됩니다.

취소 관찰

취소는 asyncio.CancelledError로 캐퍼빌리티에 도달합니다: wrap_* 훅의 handler() await를 통해(await handler(...) 주변에서 잡기), 또는 런의 종료 깔때기 on_run_error에서, 그 error는 BaseException입니다. 그것은 회복 지향 Exception-타입 훅 — on_tool_execute_error, on_node_run_error, on_model_request_error — 에는 도달하지 않습니다. 취소는 회복할 수 있는 그 스텝의 실패가 아니라 종료 제어 신호이기 때문입니다.

취소는 종료적입니다: 훅이 그것을 관찰하고 정리할 수는 있지만, 런을 회복하려고 결과를 반환하는 것은 동작하지 않아요 — Python 3.11+에서 런은 다음 스텝 경계에서 취소를 재주장합니다 (Python 3.10에서는 best-effort).

노드 훅 (Node hooks)

훅	시그니처	목적
`before_node_run`	`(ctx:` `RunContext, *, node:` `AgentNode) ->` `AgentNode`	실행 전에 노드 관찰 또는 교체
`after_node_run`	`(ctx:` `RunContext, *, node:` `AgentNode, result:` `NodeResult) ->` `NodeResult`	결과 수정 (다음 노드 또는 `End`)
`wrap_node_run`	`(ctx:` `RunContext, *, node:` `AgentNode, handler:` `WrapNodeRunHandler) ->` `NodeResult`	각 그래프 노드 실행을 감싸기
`on_node_run_error`	`(ctx:` `RunContext, *, node:` `AgentNode, error: Exception) ->` `NodeResult`	노드 오류 처리 (참고: error hooks)

wrap_node_run은 에이전트 그래프의 모든 노드(UserPromptNode, ModelRequestNode, CallToolsNode)마다 발동합니다. 이것을 오버라이드해 노드 전이를 관찰하거나, 스텝별 로깅을 추가하거나, 그래프 진행을 수정하세요:

노드 훅은 런이 어떻게 구동되든 발동합니다: agent.run(), agent_run.next(), 그리고 agent.iter() 위의 async for node in agent_run: 모두 같은 경로를 탑니다.

참고

agent.run_stream()은 예외입니다. 스트림 중간에 최종 출력이 발견되자마자 결과를 넘겨주므로, 그것을 만든 모델 요청은 before_node_run을 받지만 wrap_node_run이나 after_node_run은 받지 않아요. 훅이 모든 노드에 실행되어야 하는 정리·결과 재작성을 한다면 agent.run()이나 agent.iter()로 런을 구동하세요.

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import (
    AbstractCapability,
    AgentNode,
    NodeResult,
    WrapNodeRunHandler,
)


@dataclass
class NodeLogger(AbstractCapability[Any]):
    """Logs each node that executes during a run."""

    nodes: list[str] = field(default_factory=list)

    async def wrap_node_run(
        self, ctx: RunContext[Any], *, node: AgentNode[Any], handler: WrapNodeRunHandler[Any]
    ) -> NodeResult[Any]:
        self.nodes.append(type(node).__name__)
        return await handler(node)


logger = NodeLogger()
agent = Agent('openai:gpt-5.2', capabilities=[logger])
agent.run_sync('hello')
print(logger.nodes)
#> ['UserPromptNode', 'ModelRequestNode', 'CallToolsNode']

wrap_node_run으로 그래프 진행을 수정할 수도 있어요 — 예를 들어 런당 모델 요청 수를 제한:

from dataclasses import dataclass
from typing import Any

from pydantic_graph import End

from pydantic_ai import ModelRequestNode, RunContext
from pydantic_ai.capabilities import AbstractCapability, AgentNode, NodeResult, WrapNodeRunHandler
from pydantic_ai.result import FinalResult


@dataclass
class MaxModelRequests(AbstractCapability[Any]):
    """Limits the number of model requests per run by ending early."""

    max_requests: int = 5
    count: int = 0

    async def for_run(self, ctx: RunContext[Any]) -> 'MaxModelRequests':
        return MaxModelRequests(max_requests=self.max_requests)  # fresh per run

    async def wrap_node_run(
        self, ctx: RunContext[Any], *, node: AgentNode[Any], handler: WrapNodeRunHandler[Any]
    ) -> NodeResult[Any]:
        if isinstance(node, ModelRequestNode):
            self.count += 1
            if self.count > self.max_requests:
                return End(FinalResult(output='Max model requests reached'))
        return await handler(node)

Iterating Over an Agent's Graph에서 에이전트 그래프와 그 노드 타입에 대해 더 보세요.

모델 요청 훅 (Model request hooks)

훅	시그니처	목적
`before_model_request`	`(ctx:` `RunContext, request_context:` `ModelRequestContext) ->` `ModelRequestContext`	모델 호출 전에 메시지·설정·파라미터·모델 수정
`after_model_request`	`(ctx:` `RunContext, *, request_context:` `ModelRequestContext, response:` `ModelResponse) ->` `ModelResponse`	모델의 응답 수정
`wrap_model_request`	`(ctx:` `RunContext, *, request_context:` `ModelRequestContext, handler:` `WrapModelRequestHandler) ->` `ModelResponse`	모델 호출 감싸기
`on_model_request_error`	`(ctx:` `RunContext, *, request_context:` `ModelRequestContext, error: Exception) ->` `ModelResponse`	모델 요청 오류 처리 (참고: error hooks)

ModelRequestContext은 model, messages, model_settings, model_request_parameters를 하나의 객체로 묶어 시그니처를 미래 대비하게 합니다. 주어진 요청에 모델을 바꾸려면 request_context.model을 다른 Model 인스턴스로 설정하세요. 받은 컨텍스트를 변형하거나 그 dataclasses.replace() 복사본을 반환하세요 — 어느 쪽이든 model_id와 streaming이 이어집니다.

모델 호출을 완전히 건너뛰고 대체 응답을 제공하려면 before_model_request나 wrap_model_request에서 SkipModelRequest(response)를 발생시키세요.

before_model_request 훅은 agent.run()에 전달된 메시지 히스토리를 포함한 request_context.messages 전체 목록을 보고 수정할 수 있어요.

요청의 지침을 바꾸려면 request_context.model_request_parameters.instruction_parts를 다시 쓰세요. 그것이 모델에 보내지는 것이고, 훅이 실행된 후 메시지 히스토리에 기록된 ModelRequest가 그것에서 다시 렌더링되므로, 히스토리와 트레이스가 실제로 보낸 것을 보여줍니다. 그 메시지의 instructions에 할당하는 것은 파트에 다시 전파되지 않고 모델에 도달하지 않아요.

Skip과 체인 동작

모든 skip 예외(SkipModelRequest, SkipToolValidation, SkipToolExecution)는 훅 체인을 단락시킵니다: 남은 캐퍼빌리티의 before_* 훅은 발동하지 않고, skip된 연산에 대해 after_* 훅은 호출되지 않아요. wrap_*에서 발생한 skip은 즉시 전파됩니다 — 내부 캐퍼빌리티의 wrap 훅은 결코 실행되지 않아요.

도구 훅 (Tool hooks)

도구 처리는 두 단계가 있습니다: 검증(모델의 JSON 인수를 도구의 스키마에 대해 파싱·검증)과 실행(도구 함수 실행). 각 단계에 자체 훅이 있어요.

모든 도구 훅은 ToolDefinition을 가진 tool_def 파라미터를 받습니다.

검증 훅 — args는 검증 전 모델의 원시 str | dict[str, Any], 또는 검증 후의 유효 dict[str, Any]:

훅	시그니처	목적
`before_tool_validate`	`(ctx, *, call, tool_def, args: RawToolArgs) -> RawToolArgs`	검증 전에 원시 args 수정 (예: JSON 복구)
`after_tool_validate`	`(ctx, *, call, tool_def, args: ValidatedToolArgs) -> ValidatedToolArgs`	검증된 args 수정
`wrap_tool_validate`	`(ctx, *, call, tool_def, args: RawToolArgs, handler) -> ValidatedToolArgs`	검증 단계 감싸기
`on_tool_validate_error`	`(ctx, *, call, tool_def, args: RawToolArgs, error: ValidationError` \\ `ModelRetry) -> ValidatedToolArgs`	검증 오류 처리

before_tool_validate나 wrap_tool_validate에서 SkipToolValidation(args)를 발생시켜 검증을 건너뛰고 사전 검증된 args를 제공하세요.

도구 호출은 인수가 검증된 후에만 지연될 수 있습니다. 지연을 해결하는 사람이 그 인수를 보게 되니까요. 그래서 ApprovalRequired와 CallDeferred는 after_tool_validate에서, 그리고 wrap_tool_validate의 handler()가 반환된 후에는 wrap_tool_validate에서 발생시킬 수 있습니다. before_tool_validate에서, handler() 호출 전의 wrap_tool_validate에서, 또는 (검증이 실패했기에만 실행되는) on_tool_validate_error에서 발생시키면 그 훅을 이름으로 말하는 UserError가 나요. 허용된 지연은 도구의 args_validator에서의 것과 정확히 똑같이 동작합니다: 도구가 실행되지 않고, 재시도 예산은 손대지 않으며, 호출은 런의 DeferredToolRequests에 합류합니다.

after_tool_validate는 검증된 인수에 대한 신뢰할 수 있는 게이트로 남습니다. args_validator나 wrap_tool_validate가 이미 호출을 지연했어도 실행되므로, 거기서 거부하면(ModelRetry나 ToolFailed로) 그 지연보다 이기고, 거기서 지연하면 그것을 대체하며, 그것이 반환하는 args가 지연된 호출이 지니는 것입니다.

실행 훅 — args는 항상 유효 dict[str, Any]:

훅	시그니처	목적
`before_tool_execute`	`(ctx, *, call, tool_def, args: ValidatedToolArgs) -> ValidatedToolArgs`	실행 전에 args 수정
`after_tool_execute`	`(ctx, *, call, tool_def, args: ValidatedToolArgs, result: Any) -> Any`	실행 결과 수정
`wrap_tool_execute`	`(ctx, *, call, tool_def, args: ValidatedToolArgs, handler) -> Any`	실행 감싸기
`on_tool_execute_error`	`(ctx, *, call, tool_def, args: ValidatedToolArgs, error: Exception) -> Any`	실행 오류 처리 (참고: error hooks)

실행을 건너뛰고 대체 결과를 제공하려면 before_tool_execute나 wrap_tool_execute에서 SkipToolExecution(result)를 발생시키세요.

어떤 실행 훅이든 호출을 지연시킬 수 있지만, 도구 함수가 실행되지 않도록 before_tool_execute(또는 handler() 호출 전의 wrap_tool_execute)에서 ApprovalRequired/CallDeferred를 발생시키세요. after_tool_execute에서나 handler()가 반환된 후의 wrap_tool_execute에서의 지연은 받아들여지지만, 도구가 이미 실행됐으므로 그 사이드 이펙트가 일어났고 그 결과는 버려집니다.

도구 검증·실행 훅은 재시도를 요청하려면 ModelRetry를, 재시도 없이 실패한 도구 결과를 보고하려면 ToolFailed를 발생시킬 수 있습니다. 전체 패턴은 retries and tool failures 유발을 보세요.

출력 훅 (Output hooks)

도구 처리처럼, 출력 처리도 두 단계가 있습니다: 검증(모델의 원시 출력을 출력 스키마에 대해 파싱)과 처리(값을 추출하고 출력 함수 호출). 각 단계에 자체 훅이 있어요.

모든 출력 훅은 OutputContext(모드, 출력 타입, 스키마 정보, tool 출력용 도구 호출 세부 사항)를 가진 output_context 파라미터를 받습니다.

검증 훅은 파싱이 필요한 구조화 출력(프롬프트, 네이티브, 도구, 유니온 출력)에만 발동합니다. 평문 텍스트나 이미지 출력에는 발동하지 않아요. 처리 훅은 텍스트·구조화·이미지 출력을 포함한 모든 출력 타입에 발동합니다. tool 출력의 경우 출력 훅만 발동합니다 — 도구 훅은 완전히 건너뜁니다.

검증 훅 — 구조화 출력에만 발동; output은 str(원시 텍스트) 또는 dict(도구 args):

훅	시그니처	목적
`before_output_validate`	`(ctx, *, output_context, output: RawOutput) -> RawOutput`	검증 전에 원시 출력 수정 (예: JSON 복구)
`after_output_validate`	`(ctx, *, output_context, output: Any) -> Any`	검증된 출력 수정
`wrap_output_validate`	`(ctx, *, output_context, output: RawOutput, handler) -> Any`	검증 단계 감싸기
`on_output_validate_error`	`(ctx, *, output_context, output: RawOutput, error: ValidationError` \\ `ModelRetry) -> Any`	검증 오류 처리

처리 훅 — 모든 출력 타입에 발동; output은 유효/원시 출력. 출력 검증자(@agent.output_validator)는 처리 파이프라인 안에서(wrap_output_process 내에서) 실행되므로 after_output_process는 완전히 검증된 결과를 봅니다:

훅	시그니처	목적
`before_output_process`	`(ctx, *, output_context, output: Any) -> Any`	처리 전에 출력 수정
`after_output_process`	`(ctx, *, output_context, output: Any) -> Any`	처리된 결과 수정
`wrap_output_process`	`(ctx, *, output_context, output: Any, handler) -> Any`	처리 감싸기
`on_output_process_error`	`(ctx, *, output_context, output: Any, error: Exception) -> Any`	처리 오류 처리 (참고: error hooks)

출력 검증·처리 훅은 커스텀 메시지로 모델에 다시 시도하라고 요청하는 ModelRetry를 발생시킬 수 있습니다 — 출력 함수와 출력 검증자에서 쓰는 패턴과 같아요. 전체 패턴은 ModelRetry로 재시도 유발을 보세요.

도구 준비 (Tool preparation)

캐퍼빌리티는 두 훅으로 각 스텝에서 모델이 보는 도구 정의를 필터링하거나 수정할 수 있어요:

prepare_tools — 함수 도구만 받습니다. 모델이 직접 호출할 수 있는 도구의 필터링·수정에 쓰세요.
prepare_output_tools — 출력 도구만 받으며, ctx.retry/ctx.max_retries가 에이전트 재시도 예산의 출력 쪽을 반영해 출력 훅 라이프사이클과 일치합니다.

두 훅 모두 툴셋 수준에서 동작합니다 — 결과가 모델의 요청 파라미터와 ToolManager.tools 둘 다로 흐르므로, 필터링은 도구 실행도 차단합니다.

지연 캐퍼빌리티에서

prepare_tools는 캐퍼빌리티가 로드된 후에만 실행되고, 그다음 모든 함수 도구를 항상 켜진 캐퍼빌리티처럼 받습니다. 그 전에는 그것이 규율할 것이 없어요. 로드되지 않은 캐퍼빌리티의 도구는 모델에 광고되지도, 호출 가능하지도 않습니다.

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, RunContext, ToolDefinition
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class HideDangerousTools(AbstractCapability[Any]):
    """Hides tools matching certain name prefixes from the model."""

    hidden_prefixes: tuple[str, ...] = ('delete_', 'drop_')

    async def prepare_tools(
        self, ctx: RunContext[Any], tool_defs: list[ToolDefinition]
    ) -> list[ToolDefinition]:
        return [td for td in tool_defs if not any(td.name.startswith(p) for p in self.hidden_prefixes)]


agent = Agent('openai:gpt-5.2', capabilities=[HideDangerousTools()])


@agent.tool_plain
def delete_file(path: str) -> str:
    """Delete a file."""
    return f'deleted {path}'


@agent.tool_plain
def read_file(path: str) -> str:
    """Read a file."""
    return f'contents of {path}'


result = agent.run_sync('hello')
# The model only sees `read_file`, not `delete_file`

단순한 경우에는 내장 PrepareTools / PrepareOutputTools 캐퍼빌리티가 커스텀 서브클래스 없이 callable을 감쌉니다.

이벤트 스트림 훅 (Event stream hook)

이벤트 스트리밍(run_stream_events, event_stream_handler, UI 이벤트 스트림)이 있는 런의 경우, 캐퍼빌리티가 이벤트 스트림을 관찰하거나 변환할 수 있어요:

훅	시그니처	목적
`wrap_run_event_stream`	`(ctx:` `RunContext, *, stream: AsyncIterable[AgentStreamEvent]) -> AsyncIterable[AgentStreamEvent]`	스트리밍 이벤트 관찰·필터·변환
`on_event`	async `(self, ctx:` `RunContext, event: EventT) -> None` 메서드에 `@on_event(*event_types)`	스트림의 나머지를 건드리지 않고 선택 이벤트에 반응

이 훅은 스트림이 만들어지는 곳에서 그것을 감싸므로 모든 구동 모드에서 발동합니다: agent.run()(이 훅이 등록되면 자동으로 스트리밍 활성화), agent.run_stream(), agent.iter() — async for node in agent_run:로 advance하든, agent_run.next()로 하든, 노드를 직접 스트리밍하든. 캐퍼빌리티가 버리거나 추가한 이벤트는 수동 node.stream() 소비자가 보는 것에 반영됩니다 — 다른 소비자와 동일하게요. 그것은 또한 realtime 세션의 이벤트 이터레이터를 감싸는데, 거기서 스트림은 realtime 전용 RealtimeEvent 멤버를 추가로 포함합니다.

소비자가 이벤트 스트림을 다 소진하기 전에 닫으면, Pydantic AI는 wrap_run_event_stream이 반환한 각 래퍼도, 그것이 aclose() 메서드를 제공한다면 닫습니다. 커스텀 래퍼는 정리에 try/finally를 쓰고 거기서 안전하게 정리 코드를 await할 수 있지만, GeneratorExit를 다루는 동안 이벤트를 yield하면 안 됩니다 — 소비자가 사라졌으니까요.

자신의 입력을 닫는 래퍼와 만든 모든 래퍼를 닫는 조합 캐퍼빌리티가 같은 스트림에 도달할 수 있으므로, aclose()가 두 번 이상 호출될 수 있어요. Async generator는 여기서 멱등이라 try/finally 래퍼는 추가로 할 게 없고, aclose()를 손으로 구현하는 래퍼는 반복 호출을 no-op으로 만들어야 해요.

프로세서는 핸들러의 보기만이 아니라 전체 스트림을 형성합니다

런에는 이벤트 스트림이 하나이므로, 이벤트를 버리거나 다시 쓰는 캐퍼빌리티는 모든 소비자가 보는 것을 바꿉니다 — agent.run_stream() 호출자가 stream_text()로 얻는 텍스트를 포함해서요.

일부 이벤트는 제어 신호이기도 해요. FinalResultEvent는 agent.run_stream()에게 최종 출력이 시작됐다고 알리므로, 그것을 버리는 프로세서는 run_stream()이 결과를 스트리밍하는 대신 전체 모델 응답을 기다리게 만듭니다. 런의 출력은 바뀌지 않습니다. 의도적으로 필터링하세요.

그것은 런의 출력에 도달하지 않습니다. ModelResponse는 프로세서가 이벤트를 보기 전에 원시 모델 스트림에서 누적되므로, stream_output()와 최종 검증된 출력은 영향받지 않습니다 — 이벤트를 버리는 것은 부분 스냅샷이 언제 발행되는지만 바꾸지, 그것이 무엇을 담는지는 바꾸지 않아요. 아무것도 바꾸지 않고 관찰하려면 스트림을 받아 각 이벤트를 그대로 yield하세요.

from collections.abc import AsyncIterable
from dataclasses import dataclass
from typing import Any

from pydantic_ai import AgentStreamEvent, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.messages import (
    PartStartEvent,
    TextPart,
    ToolCallEvent,
    ToolResultEvent,
)


@dataclass
class StreamAuditor(AbstractCapability[Any]):
    """Logs tool calls and text output during streamed runs."""

    async def wrap_run_event_stream(
        self,
        ctx: RunContext[Any],
        *,
        stream: AsyncIterable[AgentStreamEvent],
    ) -> AsyncIterable[AgentStreamEvent]:
        async for event in stream:
            if isinstance(event, ToolCallEvent):
                print(f'Tool called: {event.part.tool_name}')
            elif isinstance(event, ToolResultEvent):
                print(f'Tool result: {event.part.content!r}')
            elif isinstance(event, PartStartEvent) and isinstance(event.part, TextPart):
                print(f'Text: {event.part.content!r}')
            yield event

ToolCallEvent와 ToolResultEvent에 대한 매칭은 함수 도구 호출(FunctionToolCallEvent / FunctionToolResultEvent)과 출력 도구 호출(OutputToolCallEvent / OutputToolResultEvent)을 모두 처리합니다. 다르게 취급해야 할 때는 특정 서브클래스를 매칭하세요. 지연 도구 호출은 추가로 배치 수준 DeferredToolRequestsEvent / DeferredToolResultsEvent를 발행합니다.

전체 스트림을 형성하기보다 몇 가지 이벤트 타입에 반응하려면 async 메서드를 @on_event로 표시하세요. 그것은 스트림의 같은 지점에서 이벤트를 받고, 전달한 클래스에 대한 isinstance로 필터하며, 다른 소비자가 보는 것을 바꿀 수 없어요. 클래스 이름을 지정하는 것은 또한 여러분의 캐퍼빌리티를 다른 모든 것에 대한 디스패치에서 빼줍니다 — 참고: Reacting to events.

여러분의 캐퍼빌리티는 또한 다른 캐퍼빌리티와 호스트 애플리케이션이 반응할 자체 이벤트를, 네임스페이스된 CapabilityEvent 서브클래스를 정의하고 훅이나 기여한 도구에서 ctx.emit()을 await해서 발행할 수 있어요. 양쪽 모두 Capability events를, 캐퍼빌리티가 왜 애플리케이션 CustomEvent 대신 이들을 발행하는지는 Which event type do I use?를 보세요.

스트리밍 이벤트를 프로토콜 특정 형식(SSE 같은)으로 변환하는 웹 UI를 만들려면 UI event streams 문서와 UIEventStream 베이스 클래스를 보세요.

오류 훅 (Error hooks)

각 라이프사이클 지점에는 on_*_error 훅이 있습니다 — after_*의 오류 대응입니다. after_* 훅이 성공 시 발동하는 반면, on_*_error 훅은 실패 시(wrap_*이 복구 기회를 가진 후) 발동합니다:

before_X → wrap_X(handler)
  ├─ success ─────────→ after_X (modify result)
  └─ failure → on_X_error
        ├─ re-raise ──→ (error propagates, after_X not called)
        └─ recover ───→ after_X (modify recovered result)

오류 훅은 raise-to-propagate, return-to-recover 시맨틱을 사용합니다:

원래 오류를 발생 — 오류를 그대로 전파 (기본)
다른 예외를 발생 — 오류를 변환
결과 반환 — 오류를 억제하고 반환된 값 사용

훅	발동 조건	회복 타입
`on_run_error`	에이전트 런 실패	`AgentRunResult` 반환
`on_node_run_error`	그래프 노드 실패	다음 노드 또는 `End` 반환
`on_model_request_error`	모델 요청 실패	`ModelResponse` 반환
`on_tool_validate_error`	도구 검증 실패	유효 args `dict` 반환
`on_tool_execute_error`	도구 실행 실패	어떤 도구 결과든 반환
`on_output_validate_error`	출력 검증 실패	유효 출력 반환
`on_output_process_error`	출력 실행 실패	어떤 출력 결과든 반환

여러 캐퍼빌리티에서 on_*_error 훅은 역순(after_*처럼)으로 발동합니다. 결과를 반환하는 첫 캐퍼빌리티가 오류를 회복합니다 — 남은 캐퍼빌리티의 오류 훅은 호출되지 않아요. 핸들러가 재발생하거나 새 예외를 발생시키면, 체인의 다음 캐퍼빌리티가 그 예외를 봅니다.

from dataclasses import dataclass, field
from typing import Any

from pydantic_ai import ModelRequestContext, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.messages import ModelResponse, TextPart


@dataclass
class ErrorLogger(AbstractCapability[Any]):
    """Logs all errors that occur during agent runs."""

    errors: list[str] = field(default_factory=list)

    async def on_model_request_error(
        self, ctx: RunContext[Any], *, request_context: ModelRequestContext, error: Exception
    ) -> ModelResponse:
        self.errors.append(f'Model error: {error}')
        # Return a fallback response to recover
        return ModelResponse(parts=[TextPart(content='Service temporarily unavailable.')])

    async def on_tool_execute_error(
        self, ctx: RunContext[Any], *, call: Any, tool_def: Any, args: dict[str, Any], error: Exception
    ) -> Any:
        self.errors.append(f'Tool {call.tool_name} failed: {error}')
        raise error  # Re-raise to let the normal retry flow handle it

지연 도구 호출 (Deferred tool calls)

캐퍼빌리티는 지연 도구 호출 — 승인이 필요한, 또는 외부에서 실행되는 호출 — 을 런을 끝내고 후속을 기다리지 않고 에이전트 런에서 직접 해결할 수 있어요:

훅	시그니처	목적
`handle_deferred_tool_calls`	`(ctx: RunContext, *, requests: DeferredToolRequests) -> DeferredToolResults	None`

여러 캐퍼빌리티가 각각 부분집합을 처리할 수 있어요. 디스패치는 체인을 가로질러 결과를 누적해, 아직 해결되지 않은 요청만 다음 캐퍼빌리티에 전달합니다. None(또는 항목이 없는 DeferredToolResults)을 반환하면 처리를 거절합니다. 아직 해결되지 않은 것은 호출자가 처리할 DeferredToolRequests 출력으로 올라갑니다.

핸들러를 꽂기만 하면 되는 애플리케이션 코드는 전용 HandleDeferredToolCalls 캐퍼빌리티를 쓰세요 — 참고: Resolving deferred calls with a handler.

지속 캐퍼빌리티 연산 (Durable capability operations)

캐퍼빌리티는 durable_operation으로 워크플로 코드의 I/O·비결정적 작업을 지속 activity, step, task로 옮길 수 있어요. 캐퍼빌리티에 안정적인 id를 주세요. 엔진이 캐퍼빌리티 ID와 연산 이름으로 영속된 작업을 회복·재생하기 때문입니다.

런타임 통합은 모델·도구 연산과 같은 타입 있는 백엔드를 통해 이 연산들을 받습니다. 엔진을 구현할 때는 지속 실행 백엔드 만들기를 보세요.

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import AbstractCapability, durable_operation
from pydantic_ai.models.test import TestModel


class Summaries(AbstractCapability[None]):
    id = 'summaries'

    async def before_run(self, ctx: RunContext[None]) -> None:
        summary = await self.summarize(ctx, ['one', 'two'])
        assert summary == '2 messages'

    @durable_operation(name='summarize')
    async def summarize(self, ctx: RunContext[None], messages: list[str]) -> str:
        return f'{len(messages)} messages'


agent = Agent(TestModel(), capabilities=[Summaries()])

각 연산 메서드를 @durable_operation(name='...')으로 표시하세요. 필수 이름은 영속된 지속 단위 이름의 일부가 되어 안정적으로 유지되어야 하며, 파이썬 메서드는 자유롭게 이름을 바꿀 수 있어요. 지속성 캐퍼빌리티가 바인딩되면, 런 중 메서드 호출이 그 엔진을 통해 디스패치됩니다. 지속성이 없으면 같은 호출이 원래 메서드를 직접 await합니다.

for_run 오버라이드가 새 인스턴스를 반환할 수 있습니다 — 연산은 before_run과 요청별 훅 모두에서 런이 사용하는 인스턴스에서 디스패치됩니다. 교체는 캐퍼빌리티의 id를 유지해야 합니다. 디스패치와 워커 측 회복이 그것으로 해결하기 때문이에요. Pydantic AI는 런 시작 시 바인딩된 캐퍼빌리티의 ID가 더 이상 없으면 UserError를 발생시킵니다. 디스패치는 for_run()이 반환한 후에 확립되므로, for_run() 자체 안에서 호출된 연산은 지속적으로가 아니라 직접 실행됩니다.

인수와 결과는 지속 도구와 같은 직렬화 규칙을 따라야 해요. Temporal은 그것들을 데이터 컨버터를 통해 보내고, JSON 저널 엔진은 JSON 호환 값을 요구합니다. 연산 이름은 캐퍼빌리티 ID로 범위가 정해집니다. 어느 식별자를 바꾸든 다른 영속 연산이 만들어지고, Prefect에서는 다른 캐시 키도 만들어집니다.

실시간 값 훅인 get_toolset, get_wrapper_toolset, wrap_run, wrap_node_run, wrap_model_request, wrap_tool_validate, wrap_tool_execute, wrap_output_validate, wrap_output_process, wrap_run_event_stream은 그 핸들러나 값이 지속 경계를 넘을 수 없어서 장식할 수 없습니다. Pydantic AI는 에이전트 생성 중 호환되지 않는 훅을 이름으로 말하는 UserError를 발생시킵니다.

캐퍼빌리티 감싸기 (Wrapping capabilities)

WrapperCapability는 다른 캐퍼빌리티를 감싸고 모든 메서드를 위임합니다 — 툴셋의 WrapperToolset과 유사해요. 그것을 서브클래스화해 특정 메서드를 오버라이드하고 나머지는 위임하세요:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import ModelRequestContext, RunContext
from pydantic_ai.capabilities import WrapperCapability


@dataclass
class AuditedCapability(WrapperCapability[Any]):
    """Wraps any capability and logs its model requests."""

    async def before_model_request(
        self, ctx: RunContext[Any], request_context: ModelRequestContext
    ) -> ModelRequestContext:
        print(f'Request from {type(self.wrapped).__name__}')
        return await super().before_model_request(ctx, request_context)

내장 PrefixTools는 WrapperCapability의 예입니다 — 다른 캐퍼빌리티를 감싸 그 도구 이름에 접두사를 붙여요.

런별 상태 격리 (Per-run state isolation)

생성 시 for_agent() 바인딩 후, 결과 캐퍼빌리티 인스턴스는 에이전트의 모든 런에서 공유됩니다. 캐퍼빌리티가 런 사이에 새어서는 안 되는 가변 상태를 누적한다면, for_run을 오버라이드해 새 인스턴스를 반환하세요:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class RequestCounter(AbstractCapability[Any]):
    """Counts model requests per run."""

    count: int = 0

    async def for_run(self, ctx: RunContext[Any]) -> 'RequestCounter':
        return RequestCounter()  # fresh instance for each run

    async def before_model_request(
        self, ctx: RunContext[Any], request_context: ModelRequestContext
    ) -> ModelRequestContext:
        self.count += 1
        return request_context


counter = RequestCounter()
agent = Agent('openai:gpt-5.2', capabilities=[counter])

# The shared counter stays at 0 because for_run returns a fresh instance
agent.run_sync('first run')
agent.run_sync('second run')
print(counter.count)
#> 0

for_run이 새 인스턴스를 반환하면, 캐퍼빌리티의 구성이 런 설정 시 그 교체에서 재추출됩니다: get_instructions, get_toolset, get_native_tools, get_model_settings가 그것 위에서 재호출되고, get_wrapper_toolset, get_description, 모든 라이프사이클 훅은 항상 그것 위에서 실행됩니다. 예외는 모델 선택입니다: get_model()과 부트스트랩 resolve_model_id()은 for_run 전에 원래 인스턴스에서 실행되고, 교체가 실제로 모델 기여를 바꾸지 않으면 부트스트랩 선택이 재사용됩니다 — 참고: Model selection lifecycle and limitations.

for_run 안에서 self를 변형하지 마세요 — 대신 새 인스턴스를 반환하세요. for_run이 원본을 변경 없이 반환하면 에이전트 생성 시 캐시된 구성이 재사용되므로, self에 대한 변형이 반영되지 않을 거예요.

캐퍼빌리티 동적 만들기 (Dynamically building a capability)

캐퍼빌리티는 각 에이전트 런 전에, 에이전트 RunContext를 받아 캐퍼빌리티 또는 None을 반환하는 함수를 사용해 동적으로 만들 수 있어요. 이것은 캐퍼빌리티 — 그 지침, 모델 설정, 훅, 기여한 툴셋 — 가 런 특정 정보(런의 의존성 같은)에 의존할 때 유용합니다.

동적 캐퍼빌리티를 등록하려면 RunContext를 받는 함수를 Agent 생성자나 agent.run()의 capabilities 인수에 전달하세요. sync와 async 함수 모두 지원됩니다. 이 함수는 런당 한 번 호출되고 반환된 캐퍼빌리티가 런의 나머지 동안 그것을 대체하므로, 그 지침·모델 설정·툴셋·네이티브 도구·훅이 모두 정상적으로 흐릅니다.

from dataclasses import dataclass
from typing import Literal

from pydantic_ai import Agent, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.models.test import TestModel


@dataclass
class Skill(AbstractCapability[str]):
    """Per-user skill loaded from a database at run time."""

    name: str
    role: Literal['admin', 'guest']

    def get_instructions(self) -> str:
        return f'You can use the {self.name} skill (role: {self.role}).'


# Pretend this comes from a database keyed by user.
SKILLS = {
    'alice': Skill(name='refunds', role='admin'),
    'bob': Skill(name='lookup', role='guest'),
}


def user_skill(ctx: RunContext[str]) -> AbstractCapability[str] | None:
    return SKILLS.get(ctx.deps)


agent = Agent(TestModel(), deps_type=str, capabilities=[user_skill])

result = agent.run_sync('hi', deps='alice')
print(result.all_messages()[0].instructions)
#> You can use the refunds skill (role: admin).

(이 예제는 완전하며, "그대로" 실행할 수 있어요)

단일 팩토리에서 둘 이상의 캐퍼빌리티를 반환하려면 그것들을 CombinedCapability로 감싸세요.

지속 실행 (Temporal, DBOS, Prefect)

동적 캐퍼빌리티는 기여한 툴셋을 포함해 지속 실행과 동작합니다. 세 엔진 모두에서 DynamicCapability에 안정적인 id를 설정해 그 activity·step·task가 일관되게 등록되게 하세요. capabilities=에 직접 전달된 베어 CapabilityFunc는 id를 운반할 수 없으므로 명시적으로 감쌉니다: DynamicCapability(my_func, id='...').

DBOS는 동적 도구 발견·호출을 스텝에서 실행해 MCP I/O를 체크포인트하고, Prefect는 동적 도구 호출을 태스크에서, 도구 해석을 플로우 코드에서 실행합니다. 둘 다 그 스텝·태스크 안에서 런을 위해 이미 해석된 캐퍼빌리티를 재사용합니다. Temporal은 activity 경계가 런의 해석된 캐퍼빌리티를 운반할 수 없으므로 액티비티 안에서 팩토리를 다시 실행합니다. 세 엔진 모두에서 팩토리 자체는 워크플로·플로우 코드에서 실행되어 재생·회복·플로우 재시도 시 다시 실행되므로 — 런 의존성에 대해 결정적으로 유지하고 I/O는 그것이 반환하는 툴셋에 맡기세요.

조립과 미들웨어 시맨틱 (Composition and middleware semantics)

에이전트에 여러 캐퍼빌리티가 전달되면, 그것들은 미들웨어 시맨틱을 따르는 단일 CombinedCapability으로 조립됩니다 — Django나 Starlette 같은 웹 프레임워크가 쓰는 패턴과 같아요:

구성이 병합됩니다: 지침은 연결되고, 모델 설정은 가산적으로 병합되며(나중 캐퍼빌리티가 이른 것을 덮어씀), 툴셋은 결합되고, 네이티브 도구는 모입니다.
before_* 훅은 캐퍼빌리티 순서로(바깥에서 안으로) 발동합니다: cap1 → cap2 → cap3.
after_* 훅은 역순으로(안에서 바깥으로) 발동합니다: cap3 → cap2 → cap1.
wrap_* 훅은 미들웨어로 중첩됩니다: cap1이 cap2를 감싸고 cap2가 cap3을 감싸고 cap3이 실제 연산을 감쌉니다. 첫 캐퍼빌리티가 가장 바깥 계층입니다.
**get_wrapper_toolset**도 같은 중첩을 따릅니다: 첫 캐퍼빌리티의 래퍼가 가장 바깥입니다.

이것은 목록의 첫 캐퍼빌리티가 연산에 대해 첫 번째와 마지막 발언권을 가짐을 뜻합니다 — 그것은 다른 어떤 캐퍼빌리티보다 먼저 원래 입력을 보고, 모든 내부 캐퍼빌리티가 처리한 후 최종 출력을 봅니다.

순서 (Ordering)

기본적으로 캐퍼빌리티는 나열한 순서로 조립됩니다. 캐퍼빌리티가 사용자가 어디에 나열하든 특정 위치에 있어야 할 때는 get_ordering을 오버라이드해 CapabilityOrdering을 반환하세요:

from dataclasses import dataclass
from typing import Any

from pydantic_ai.capabilities import (
    AbstractCapability,
    CapabilityOrdering,
    CombinedCapability,
)


@dataclass
class InstrumentationCapability(AbstractCapability[Any]):
    """Must wrap all other capabilities to trace everything."""

    def get_ordering(self) -> CapabilityOrdering:
        return CapabilityOrdering(position='outermost')


@dataclass
class PlainCapability(AbstractCapability[Any]):
    pass


# InstrumentationCapability ends up first regardless of list order
combined = CombinedCapability([PlainCapability(), InstrumentationCapability()])
assert type(combined.capabilities[0]) is InstrumentationCapability

사용 가능한 제약:

position — 'outermost' 또는 'innermost'. 캐퍼빌리티를 그 위치가 없는 모든 캐퍼빌리티보다 앞(또는 뒤)의 계층에 둡니다. 여러 캐퍼빌리티가 계층을 공유할 수 있고, 원래 목록 순서가 그 안에서 동률을 깨요.
wraps — 이 캐퍼빌리티가(바깥에서) 감싸는 캐퍼빌리티 목록. 각 항목은 캐퍼빌리티 타입(모든 인스턴스 issubclass 매칭) 또는 특정 인스턴스(동일성 매칭)일 수 있어요. 캐퍼빌리티가 다른 것의 출력을 봐야 할 때 쓰세요: CapabilityOrdering(wraps=[OtherCapability]).
wrapped_by — 이 캐퍼빌리티를(바깥에서) 감싸는 캐퍼빌리티 목록. wraps처럼 타입이나 인스턴스를 받습니다. wraps의 역입니다.
requires — 반드시 있어야 하는 캐퍼빌리티 타입 목록. 하나라도 없으면 UserError를 발생시킵니다. 순서를 암시하지 않아요.

제약이 선언되면 CombinedCapability은 생성 시 자식을 위상 정렬하며, 사용자 제공 순서를 동률 해결책으로 유지합니다.

Hooks는 ordering 파라미터로 순서를 지원하므로, 서브클래싱 없이 순서 제약을 선언할 수 있어요:

from pydantic_ai.capabilities import CapabilityOrdering, CombinedCapability, Hooks

logging_hooks = Hooks(ordering=CapabilityOrdering(position='outermost'))
rate_limit_hooks = Hooks(ordering=CapabilityOrdering(wrapped_by=[logging_hooks]))

# logging_hooks ends up outermost; rate_limit_hooks is wrapped by it
combined = CombinedCapability([rate_limit_hooks, logging_hooks])
assert combined.capabilities[0] is logging_hooks
assert combined.capabilities[1] is rate_limit_hooks

캐퍼빌리티는 서로 직접 접근하지 않습니다. 다른 캐퍼빌리티에게 무언가 일어났다고 알리려면 캐퍼빌리티 이벤트를 발행하세요. 발행자는 ctx.emit()을 await하고, 관심 있는 어떤 캐퍼빌리티든 @on_event로 반응합니다. 그 사이에 공유 객체도, capabilities 목록의 순서 요구도 없어요.

발생을 알리는 대신 값을 공유하려면 async 함수에서 설정하는 contextvars.ContextVar를 쓰세요: 한 캐퍼빌리티가 설정하고(예: wrap_run이나 before_run), 다른 하나가 그 훅에서 읽습니다. capabilities 목록의 캐퍼빌리티 순서가 중요해요 — 작성자가 읽는 자보다 앞에 있어 그 before_* 훅이 먼저 실행되게 해야 합니다. sync Hooks 함수는 작성자가 될 수 없어요. 별도 스레드에서 실행되므로 그것이 설정한 값이 런의 나머지에 보이지 않습니다.

커스텀 캐퍼빌리티 테스트 (Testing custom capabilities)

커스텀 캐퍼빌리티를 에이전트 테스트와 같은 방식으로 테스트하세요 — TestModel이나 FunctionModel을 씁니다. 캐퍼빌리티로 에이전트를 만들고 런 결과, 메시지, 또는 훅의 관찰 가능한 사이드 이펙트에 대해 단언하세요.

예제 (Examples)

가드레일 (Guardrail, PII 삭제)

가드레일은 안전 규칙을 강제하기 위해 모델 요청·응답을 가로채는 캐퍼빌리티예요. 다음은 모델 응답에서 잠재적 PII를 스캔해 삭제하는 것입니다:

import re
from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, ModelRequestContext, RunContext
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.messages import ModelResponse, TextPart


@dataclass
class PIIRedactionGuardrail(AbstractCapability[Any]):
    """Redacts email addresses and phone numbers from model responses."""

    async def after_model_request(
        self,
        ctx: RunContext[Any],
        *,
        request_context: ModelRequestContext,
        response: ModelResponse,
    ) -> ModelResponse:
        for part in response.parts:
            if isinstance(part, TextPart):
                # Redact email addresses
                part.content = re.sub(
                    r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
                    '[EMAIL REDACTED]',
                    part.content,
                )
                # Redact phone numbers (simple US pattern)
                part.content = re.sub(
                    r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
                    '[PHONE REDACTED]',
                    part.content,
                )
        return response


agent = Agent('openai:gpt-5.2', capabilities=[PIIRedactionGuardrail()])
result = agent.run_sync("What's Jane's contact info?")
print(result.output)
#> You can reach Jane at [EMAIL REDACTED] or [PHONE REDACTED].

로깅 미들웨어 (Logging middleware)

wrap_* 패턴은 연산의 입력과 출력을 모두 관찰하거나 시간을 재고 싶을 때 유용해요. 다음은 모든 모델 요청과 도구 호출을 로그하는 캐퍼빌리티입니다:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, ModelRequestContext, RunContext, ToolDefinition
from pydantic_ai.capabilities import (
    AbstractCapability,
    WrapModelRequestHandler,
    WrapToolExecuteHandler,
)
from pydantic_ai.messages import ModelResponse, ToolCallPart


@dataclass
class VerboseLogging(AbstractCapability[Any]):
    """Logs model requests and tool executions."""

    async def wrap_model_request(
        self,
        ctx: RunContext[Any],
        *,
        request_context: ModelRequestContext,
        handler: WrapModelRequestHandler,
    ) -> ModelResponse:
        print(f'  Model request (step {ctx.run_step}, {len(request_context.messages)} messages)')
        #>   Model request (step 1, 1 messages)
        response = await handler(request_context)
        print(f'  Model response: {len(response.parts)} parts')
        #>   Model response: 1 parts
        return response

    async def wrap_tool_execute(
        self,
        ctx: RunContext[Any],
        *,
        call: ToolCallPart,
        tool_def: ToolDefinition,
        args: dict[str, Any],
        handler: WrapToolExecuteHandler,
    ) -> Any:
        print(f'  Tool call: {call.tool_name}({args})')
        result = await handler(args)
        print(f'  Tool result: {result!r}')
        return result


agent = Agent('openai:gpt-5.2', capabilities=[VerboseLogging()])
result = agent.run_sync('hello')
print(f'Output: {result.output}')
#> Output: Hello! How can I help you today?

캐퍼빌리티 배포 (Publishing capabilities)

커스텀 캐퍼빌리티를 에이전트 스펙에서 사용 가능하게 하려면 get_serialization_name(클래스 이름으로 기본 설정)과 직렬화 가능 인수를 받는 생성자가 필요해요. 기본 from_spec 구현은 cls(*args, **kwargs)를 호출하므로, 단순 dataclass는 오버라이드가 필요 없습니다:

from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent, AgentSpec
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class RateLimit(AbstractCapability[Any]):
    """Limits requests per minute."""

    rpm: int = 60


# In YAML: `- RateLimit: {rpm: 30}`
# In Python:
agent = Agent.from_spec(
    AgentSpec(model='test', capabilities=[{'RateLimit': {'rpm': 30}}]),
    custom_capability_types=[RateLimit],
)

사용자는 Agent.from_spec 또는 Agent.from_file의 custom_capability_types 파라미터로 커스텀 캐퍼빌리티 타입을 등록합니다.

생성자가 YAML/JSON으로 표현할 수 없는 타입을 받을 때 from_spec을 오버라이드하세요. 스펙 필드는 dataclass 필드를 반영해야 하되 직렬화 가능 타입이어야 합니다:

from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

from pydantic_ai import RunContext, ToolDefinition
from pydantic_ai.capabilities import AbstractCapability


@dataclass
class ConditionalTools(AbstractCapability[Any]):
    """Hides tools unless a condition is met."""

    condition: Callable[[RunContext[Any]], bool]  # not serializable
    hidden_tools: list[str] = field(default_factory=list)

    @classmethod
    def from_spec(cls, hidden_tools: list[str]) -> 'ConditionalTools':
        # In the spec, there's no condition callable -- always hide
        return cls(condition=lambda ctx: True, hidden_tools=hidden_tools)

    async def prepare_tools(
        self, ctx: RunContext[Any], tool_defs: list[ToolDefinition]
    ) -> list[ToolDefinition]:
        if self.condition(ctx):
            return [td for td in tool_defs if td.name not in self.hidden_tools]
        return tool_defs

YAML에서는 - ConditionalTools: {hidden_tools: [dangerous_tool]}이 됩니다. 파이썬 코드에서는 전체 생성자가 사용 가능합니다: ConditionalTools(condition=my_check, hidden_tools=['dangerous_tool']).

패키징 규칙과 더 넓은 확장 생태계는 Extensibility를 보세요.

커스텀 캐퍼빌리티 만들기