Module adcp.types.core

Core type definitions.

Classes

class Activity (**data: Any)

Expand source code

class Activity(BaseModel):
    """Activity event for observability."""

    model_config = {"frozen": True}

    type: ActivityType
    operation_id: str
    agent_id: str
    task_type: str
    status: TaskStatus | None = None
    timestamp: str
    metadata: dict[str, Any] | None = None

Activity event for observability.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var agent_id : str

var metadata : dict[str, typing.Any] | None

var model_config

var operation_id : str

var status : TaskStatus | None

var task_type : str

var timestamp : str

var type : ActivityType

class ActivityType (*args, **kwds)

Expand source code

class ActivityType(str, Enum):
    """Types of activity events."""

    PROTOCOL_REQUEST = "protocol_request"
    PROTOCOL_RESPONSE = "protocol_response"
    WEBHOOK_RECEIVED = "webhook_received"
    HANDLER_CALLED = "handler_called"
    STATUS_CHANGE = "status_change"

Types of activity events.

Ancestors

• builtins.str
• enum.Enum

Class variables

var HANDLER_CALLED

var PROTOCOL_REQUEST

var PROTOCOL_RESPONSE

var STATUS_CHANGE

var WEBHOOK_RECEIVED

class AgentConfig (**data: Any)

Expand source code

class AgentConfig(BaseModel):
    """Agent configuration."""

    id: str
    agent_uri: str
    protocol: Protocol
    auth_token: str | None = None
    requires_auth: bool = False
    auth_header: str = "x-adcp-auth"  # Header name for authentication
    auth_type: str = "token"  # "token" for direct value, "bearer" for "Bearer {token}"
    timeout: float = 30.0  # Request timeout in seconds
    mcp_transport: str = (
        "streamable_http"  # "streamable_http" (default, modern) or "sse" (legacy fallback)
    )
    debug: bool = False  # Enable debug mode to capture request/response details

    @field_validator("agent_uri")
    @classmethod
    def validate_agent_uri(cls, v: str) -> str:
        """Validate agent URI format."""
        if not v:
            raise ValueError("agent_uri cannot be empty")

        if not v.startswith(("http://", "https://")):
            raise ValueError(
                f"agent_uri must start with http:// or https://, got: {v}\n"
                "Example: https://agent.example.com"
            )

        # Remove trailing slash for consistency
        return v.rstrip("/")

    @field_validator("timeout")
    @classmethod
    def validate_timeout(cls, v: float) -> float:
        """Validate timeout is reasonable."""
        if v <= 0:
            raise ValueError(f"timeout must be positive, got: {v}")

        if v > 300:  # 5 minutes
            raise ValueError(
                f"timeout is very large ({v}s). Consider a value under 300 seconds.\n"
                "Large timeouts can cause long hangs if agent is unresponsive."
            )

        return v

    @field_validator("mcp_transport")
    @classmethod
    def validate_mcp_transport(cls, v: str) -> str:
        """Validate MCP transport type."""
        valid_transports = ["streamable_http", "sse"]
        if v not in valid_transports:
            raise ValueError(
                f"mcp_transport must be one of {valid_transports}, got: {v}\n"
                "Use 'streamable_http' for modern agents (recommended)"
            )
        return v

    @field_validator("auth_type")
    @classmethod
    def validate_auth_type(cls, v: str) -> str:
        """Validate auth type."""
        valid_types = ["token", "bearer"]
        if v not in valid_types:
            raise ValueError(
                f"auth_type must be one of {valid_types}, got: {v}\n"
                "Use 'bearer' for OAuth2/standard Authorization header"
            )
        return v

Agent configuration.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var agent_uri : str

var auth_header : str

var auth_token : str | None

var auth_type : str

var debug : bool

var id : str

var mcp_transport : str

var model_config

var protocol : Protocol

var requires_auth : bool

var timeout : float

Static methods

def validate_agent_uri(v: str) ‑> str

Validate agent URI format.

def validate_auth_type(v: str) ‑> str

Validate auth type.

def validate_mcp_transport(v: str) ‑> str

Validate MCP transport type.

def validate_timeout(v: float) ‑> float

Validate timeout is reasonable.

class DebugInfo (**data: Any)

Expand source code

class DebugInfo(BaseModel):
    """Debug information for troubleshooting."""

    request: dict[str, Any]
    response: dict[str, Any]
    duration_ms: float | None = None

Debug information for troubleshooting.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var duration_ms : float | None

var model_config

var request : dict[str, typing.Any]

var response : dict[str, typing.Any]

class Member (**data: Any)

Expand source code

class Member(BaseModel):
    """An organization registered in the AAO member directory."""

    model_config = ConfigDict(extra="allow")

    id: str
    slug: str
    display_name: str
    description: str | None = None
    tagline: str | None = None
    logo_url: str | None = None
    logo_light_url: str | None = None
    logo_dark_url: str | None = None
    contact_email: str | None = None
    contact_website: str | None = None
    offerings: list[str] = Field(default_factory=list)
    markets: list[str] = Field(default_factory=list)
    agents: list[dict[str, Any]] = Field(default_factory=list)
    brands: list[dict[str, Any]] = Field(default_factory=list)
    is_public: bool = True
    is_founding_member: bool = False
    featured: bool = False
    si_enabled: bool = False

An organization registered in the AAO member directory.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var agents : list[dict[str, typing.Any]]

var brands : list[dict[str, typing.Any]]

var contact_email : str | None

var contact_website : str | None

var description : str | None

var display_name : str

var featured : bool

var id : str

var is_founding_member : bool

var is_public : bool

var logo_dark_url : str | None

var logo_light_url : str | None

var logo_url : str | None

var markets : list[str]

var model_config

var offerings : list[str]

var si_enabled : bool

var slug : str

var tagline : str | None

class NeedsInputInfo (**data: Any)

Expand source code

class NeedsInputInfo(BaseModel):
    """Information when agent needs clarification."""

    message: str
    field: str | None = None

Information when agent needs clarification.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var field : str | None

var message : str

var model_config

class Policy (**data: Any)

Expand source code

class Policy(PolicySummary):
    """Full governance policy including policy text and calibration exemplars."""

    policy: str
    guidance: str | None = None
    exemplars: PolicyExemplars | None = None
    ext: dict[str, Any] | None = None

Full governance policy including policy text and calibration exemplars.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• PolicySummary
• pydantic.main.BaseModel

Class variables

var exemplars : PolicyExemplars | None

var ext : dict[str, typing.Any] | None

var guidance : str | None

var model_config

var policy : str

class PolicyExemplar (**data: Any)

Expand source code

class PolicyExemplar(BaseModel):
    """A pass/fail scenario used to calibrate governance agent interpretation."""

    model_config = ConfigDict(extra="allow")

    scenario: str
    explanation: str

A pass/fail scenario used to calibrate governance agent interpretation.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var explanation : str

var model_config

var scenario : str

class PolicyExemplars (**data: Any)

Expand source code

class PolicyExemplars(BaseModel):
    """Collection of pass/fail exemplars for a policy."""

    model_config = ConfigDict(extra="allow")

    pass_: list[PolicyExemplar] = Field(default_factory=list, alias="pass")
    fail: list[PolicyExemplar] = Field(default_factory=list)

Collection of pass/fail exemplars for a policy.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var fail : list[PolicyExemplar]

var model_config

var pass_ : list[PolicyExemplar]

class PolicyHistory (**data: Any)

Expand source code

class PolicyHistory(BaseModel):
    """Edit history for a policy."""

    model_config = ConfigDict(extra="allow")

    policy_id: str
    total: int
    revisions: list[PolicyRevision] = Field(default_factory=list)

Edit history for a policy.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var model_config

var policy_id : str

var revisions : list[PolicyRevision]

var total : int

class PolicyRevision (**data: Any)

Expand source code

class PolicyRevision(BaseModel):
    """A single revision in a policy's edit history."""

    model_config = ConfigDict(extra="allow")

    revision_number: int
    editor_name: str
    edit_summary: str
    is_rollback: bool
    rolled_back_to: int | None = None
    created_at: str

A single revision in a policy's edit history.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var created_at : str

var edit_summary : str

var editor_name : str

var is_rollback : bool

var model_config

var revision_number : int

var rolled_back_to : int | None

class PolicySummary (**data: Any)

Expand source code

class PolicySummary(BaseModel):
    """Summary of a governance policy from the registry."""

    model_config = ConfigDict(extra="allow", populate_by_name=True)

    policy_id: str
    version: str
    name: str
    description: str | None = None
    category: str
    enforcement: str
    jurisdictions: list[str] = Field(default_factory=list)
    region_aliases: dict[str, list[str]] = Field(default_factory=dict)
    verticals: list[str] = Field(default_factory=list)
    channels: list[str] | None = None
    governance_domains: list[str] = Field(default_factory=list)
    effective_date: str | None = None
    sunset_date: str | None = None
    source_url: str | None = None
    source_name: str | None = None
    source_type: str | None = None
    review_status: str | None = None
    created_at: str | None = None
    updated_at: str | None = None

Summary of a governance policy from the registry.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Subclasses

• Policy

Class variables

var category : str

var channels : list[str] | None

var created_at : str | None

var description : str | None

var effective_date : str | None

var enforcement : str

var governance_domains : list[str]

var jurisdictions : list[str]

var model_config

var name : str

var policy_id : str

var region_aliases : dict[str, list[str]]

var review_status : str | None

var source_name : str | None

var source_type : str | None

var source_url : str | None

var sunset_date : str | None

var updated_at : str | None

var version : str

var verticals : list[str]

class Protocol (*args, **kwds)

Expand source code

class Protocol(str, Enum):
    """Supported protocols."""

    A2A = "a2a"
    MCP = "mcp"

Supported protocols.

Ancestors

• builtins.str
• enum.Enum

Class variables

var A2A

var MCP

class ResolvedBrand (**data: Any)

Expand source code

class ResolvedBrand(BaseModel):
    """Brand identity resolved from the AdCP registry."""

    model_config = ConfigDict(extra="allow")

    canonical_id: str
    canonical_domain: str
    brand_name: str
    names: list[dict[str, str]] | None = None
    keller_type: str | None = None
    parent_brand: str | None = None
    house_domain: str | None = None
    house_name: str | None = None
    brand_agent_url: str | None = None
    brand: dict[str, Any] | None = None
    brand_manifest: dict[str, Any] | None = None
    source: str

    @model_validator(mode="before")
    @classmethod
    def _normalize_brand_fields(cls, data: Any) -> Any:
        """Cross-populate brand and brand_manifest so both names are always accessible."""
        if isinstance(data, dict):
            data = dict(data)
            if "brand" in data and "brand_manifest" not in data:
                data["brand_manifest"] = data["brand"]
            elif "brand_manifest" in data and "brand" not in data:
                data["brand"] = data["brand_manifest"]
        return data

Brand identity resolved from the AdCP registry.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var brand : dict[str, typing.Any] | None

var brand_agent_url : str | None

var brand_manifest : dict[str, typing.Any] | None

var brand_name : str

var canonical_domain : str

var canonical_id : str

var house_domain : str | None

var house_name : str | None

var keller_type : str | None

var model_config

var names : list[dict[str, str]] | None

var parent_brand : str | None

var source : str

class ResolvedProperty (**data: Any)

Expand source code

class ResolvedProperty(BaseModel):
    """Property information resolved from the AdCP registry."""

    model_config = ConfigDict(extra="allow")

    publisher_domain: str
    source: str
    authorized_agents: list[dict[str, Any]]
    properties: list[dict[str, Any]]
    verified: bool

Property information resolved from the AdCP registry.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var authorized_agents : list[dict[str, typing.Any]]

var model_config

var properties : list[dict[str, typing.Any]]

var publisher_domain : str

var source : str

var verified : bool

class SubmittedInfo (**data: Any)

Expand source code

class SubmittedInfo(BaseModel):
    """Information about submitted async task."""

    webhook_url: str
    operation_id: str

Information about submitted async task.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var model_config

var operation_id : str

var webhook_url : str

class TaskResult (**data: Any)

Expand source code

class TaskResult(BaseModel, Generic[T]):
    """Result from task execution."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    status: TaskStatus
    data: T | None = None
    message: str | None = None  # Human-readable message from agent (e.g., MCP content text)
    submitted: SubmittedInfo | None = None
    needs_input: NeedsInputInfo | None = None
    error: str | None = None
    success: bool = Field(default=True)
    metadata: dict[str, Any] | None = None
    debug_info: DebugInfo | None = None

Result from task execution.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel
• typing.Generic

Subclasses

• adcp.types.core.TaskResult[AdcpAsyncResponseData]
• adcp.types.core.TaskResult[Any]
• adcp.types.core.TaskResult[CheckGovernanceResponse]
• adcp.types.core.TaskResult[CreatePropertyListResponse]
• adcp.types.core.TaskResult[DeletePropertyListResponse]
• adcp.types.core.TaskResult[GetAdcpCapabilitiesResponse]
• adcp.types.core.TaskResult[GetCreativeDeliveryResponse]
• adcp.types.core.TaskResult[GetMediaBuyDeliveryResponse]
• adcp.types.core.TaskResult[GetMediaBuysResponse]
• adcp.types.core.TaskResult[GetPlanAuditLogsResponse]
• adcp.types.core.TaskResult[GetProductsResponse]
• adcp.types.core.TaskResult[GetPropertyListResponse]
• adcp.types.core.TaskResult[GetSignalsResponse]
• adcp.types.core.TaskResult[ListAccountsResponse]
• adcp.types.core.TaskResult[ListCreativeFormatsResponse]
• adcp.types.core.TaskResult[ListCreativesResponse]
• adcp.types.core.TaskResult[ListPropertyListsResponse]
• adcp.types.core.TaskResult[ReportPlanOutcomeResponse]
• adcp.types.core.TaskResult[ReportUsageResponse]
• adcp.types.core.TaskResult[SiGetOfferingResponse]
• adcp.types.core.TaskResult[SiInitiateSessionResponse]
• adcp.types.core.TaskResult[SiSendMessageResponse]
• adcp.types.core.TaskResult[SiTerminateSessionResponse]
• adcp.types.core.TaskResult[SyncPlansResponse]
• adcp.types.core.TaskResult[Union[ActivateSignalResponse1, ActivateSignalResponse2]]
• adcp.types.core.TaskResult[Union[BuildCreativeResponse1, BuildCreativeResponse2, BuildCreativeResponse3]]
• adcp.types.core.TaskResult[Union[CalibrateContentResponse1, CalibrateContentResponse2]]
• adcp.types.core.TaskResult[Union[CreateContentStandardsResponse1, CreateContentStandardsResponse2]]
• adcp.types.core.TaskResult[Union[CreateMediaBuyResponse1, CreateMediaBuyResponse2]]
• adcp.types.core.TaskResult[Union[GetAccountFinancialsResponse1, GetAccountFinancialsResponse2]]
• adcp.types.core.TaskResult[Union[GetContentStandardsResponse1, GetContentStandardsResponse2]]
• adcp.types.core.TaskResult[Union[GetCreativeFeaturesResponse1, GetCreativeFeaturesResponse2]]
• adcp.types.core.TaskResult[Union[GetMediaBuyArtifactsResponse1, GetMediaBuyArtifactsResponse2]]
• adcp.types.core.TaskResult[Union[ListContentStandardsResponse1, ListContentStandardsResponse2]]
• adcp.types.core.TaskResult[Union[LogEventResponse1, LogEventResponse2]]
• adcp.types.core.TaskResult[Union[PreviewCreativeResponse1, PreviewCreativeResponse2, PreviewCreativeResponse3]]
• adcp.types.core.TaskResult[Union[ProvidePerformanceFeedbackResponse1, ProvidePerformanceFeedbackResponse2]]
• adcp.types.core.TaskResult[Union[SyncAccountsResponse1, SyncAccountsResponse2]]
• adcp.types.core.TaskResult[Union[SyncAudiencesResponse1, SyncAudiencesResponse2]]
• adcp.types.core.TaskResult[Union[SyncCatalogsResponse1, SyncCatalogsResponse2]]
• adcp.types.core.TaskResult[Union[SyncCreativesResponse1, SyncCreativesResponse2]]
• adcp.types.core.TaskResult[Union[SyncEventSourcesResponse1, SyncEventSourcesResponse2]]
• adcp.types.core.TaskResult[Union[UpdateContentStandardsResponse1, UpdateContentStandardsResponse2]]
• adcp.types.core.TaskResult[Union[UpdateMediaBuyResponse1, UpdateMediaBuyResponse2]]
• adcp.types.core.TaskResult[Union[ValidateContentDeliveryResponse1, ValidateContentDeliveryResponse2]]
• adcp.types.core.TaskResult[UpdatePropertyListResponse]

Class variables

var data : ~T | None

var debug_info : DebugInfo | None

var error : str | None

var message : str | None

var metadata : dict[str, typing.Any] | None

var model_config

var needs_input : NeedsInputInfo | None

var status : TaskStatus

var submitted : SubmittedInfo | None

var success : bool

class TaskStatus (*args, **kwds)

Expand source code

class TaskStatus(str, Enum):
    """Task execution status."""

    COMPLETED = "completed"
    SUBMITTED = "submitted"
    NEEDS_INPUT = "needs_input"
    FAILED = "failed"
    WORKING = "working"

Task execution status.

Ancestors

• builtins.str
• enum.Enum

Class variables

var COMPLETED

var FAILED

var NEEDS_INPUT

var SUBMITTED

var WORKING

class WebhookMetadata (**data: Any)

Expand source code

class WebhookMetadata(BaseModel):
    """Metadata passed to webhook handlers."""

    operation_id: str
    agent_id: str
    task_type: str
    status: TaskStatus
    sequence_number: int | None = None
    notification_type: Literal["scheduled", "final", "delayed"] | None = None
    timestamp: str

Metadata passed to webhook handlers.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

• pydantic.main.BaseModel

Class variables

var agent_id : str

var model_config

var notification_type : Literal['scheduled', 'final', 'delayed'] | None

var operation_id : str

var sequence_number : int | None

var status : TaskStatus

var task_type : str

var timestamp : str