Module adcp.types.core

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 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 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 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[ActivateSignalResponse]
• adcp.types.core.TaskResult[Any]
• adcp.types.core.TaskResult[GetMediaBuyDeliveryResponse]
• adcp.types.core.TaskResult[GetProductsResponse]
• adcp.types.core.TaskResult[GetSignalsResponse]
• adcp.types.core.TaskResult[ListAuthorizedPropertiesResponse]
• adcp.types.core.TaskResult[ListCreativeFormatsResponse]
• adcp.types.core.TaskResult[ListCreativesResponse]
• adcp.types.core.TaskResult[PreviewCreativeResponse]
• adcp.types.core.TaskResult[ProvidePerformanceFeedbackResponse]
• adcp.types.core.TaskResult[SyncCreativesResponse]

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