​
Pebbling Protocol Types

The Pebbling Protocol defines a comprehensive set of type definitions for communication between agents and the Pebbling framework. These types ensure type-safe, structured communication across all system components.

​
Core Communication Types

​
Message Parts

The protocol supports multi-part messages with different content types:

​
TextPart

class TextPart(TypedDict):
    kind: Required[Literal["text"]]
    metadata: NotRequired[dict[str, Any]]
    text: Required[str]
    embeddings: NotRequired[list[float]]

​
FilePart

class FilePart(TextPart):
    kind: Required[Literal["file"]]
    file: Required[FileWithBytes | FileWithUri]

​
DataPart

class DataPart(TextPart):
    kind: Required[Literal["data"]]
    data: Required[dict[str, Any]]
    embeddings: NotRequired[list[float]]

​
Core Protocol Entities

​
Artifacts

Artifacts represent the final output generated by an agent after completing a task. They are immutable data structures containing the results of agent execution. 
class Artifact(TypedDict):
    artifact_id: Required[UUID]
    name: NotRequired[str]
    description: NotRequired[str]
    metadata: NotRequired[dict[str, Any]]
    parts: NotRequired[list[Part]]
    append: NotRequired[bool]
    last_chunk: NotRequired[bool]
    extra_data: NotRequired[dict[str, Any]]
Key Properties:
  • Immutable: Once created, artifacts cannot be modified
  • Multi-part: Can contain text, files, and structured data
  • Streamable: Parts can be appended during streaming responses
  • Traceable: Each artifact has a unique ID for reference

​
Messages

Messages represent all non-result communication in the Pebbling protocol, carrying operational content like instructions, status updates, and metadata. 
class Message(TypedDict):
    context_id: Required[UUID]
    task_id: Required[UUID]
    reference_task_ids: NotRequired[list[UUID]]
    message_id: Required[UUID]
    kind: Required[Literal["message"]]
    metadata: NotRequired[dict[str, Any]]
    parts: Required[list[Part]]
    role: Required[Role]
    extra_data: NotRequired[dict[str, Any]]
Message Types:
  • User Instructions: Task requests with context and files
  • Agent Communication: Status updates, thoughts, coordination
  • System Messages: Errors, warnings, protocol information
  • Context Sharing: Background information, references, metadata

​
Tasks

Tasks serve as the primary coordination mechanism, managing the complete lifecycle from request to completion. 
class Task(TypedDict):
    task_id: Required[UUID]
    context_id: Required[UUID]
    kind: Required[Literal["task"]]
    status: Required[TaskStatus]
    artifacts: NotRequired[list[Artifact]]
    history: NotRequired[list[Message]]
    metadata: NotRequired[dict[str, Any]]
Core Responsibilities:
  • Message Exchange: Facilitate communication between clients and agents
  • State Management: Track task progress and execution status
  • Artifact Collection: Gather and organize agent-generated outputs
  • History Tracking: Maintain complete conversation and decision trail
Task Lifecycle:
  1. Creation: Client initiates task with initial message/requirements
  2. Processing: Agent processes messages and updates status
  3. Communication: Bidirectional message exchange as needed
  4. Artifact Generation: Agent produces deliverable outputs
  5. Completion: Final status update and artifact delivery

​
Contexts

Contexts serve as conversation containers, managing the complete interaction lifecycle between clients and agents. 
class Context(TypedDict):
    context_id: Required[UUID]
    kind: Required[Literal["context"]]
    tasks: NotRequired[list[UUID]]
    name: NotRequired[str]
    description: NotRequired[str]
    role: Required[str]
    created_at: Required[str]
    updated_at: Required[str]
    status: NotRequired[Literal["active", "paused", "completed", "archived"]]
    tags: NotRequired[list[str]]
    metadata: NotRequired[dict[str, Any]]
    parent_context_id: NotRequired[UUID]
    reference_context_ids: NotRequired[list[UUID]]
    extra_data: NotRequired[dict[str, Any]]
Core Responsibilities:
  • Session Management: Group related tasks under a unified conversation
  • History Preservation: Maintain complete message history across tasks
  • Context Continuity: Preserve conversation state and references
  • Metadata Tracking: Store session-level information and preferences

​
File Types

​
FileWithBytes

class FileWithBytes(TypedDict):
    bytes: Required[str]
    mimeType: NotRequired[str]
    name: NotRequired[str]
    embeddings: NotRequired[list[float]]

​
FileWithUri

class FileWithUri(FileWithBytes):
    uri: Required[str]

​
Task Status

class TaskStatus(TypedDict):
    message: NotRequired[Message]
    state: Required[TaskState]
    timestamp: Required[str]  # ISO datetime value

​
Type Aliases

​
Task States

TaskState: TypeAlias = Literal[
    "submitted", "working", "input-required", "completed", 
    "canceled", "failed", "rejected", "auth-required", 
    "unknown", "trust-verification-required", "pending", 
    "suspended", "resumed", "negotiation-bid-submitted", 
    "negotiation-bid-lost", "negotiation-bid-won"
]

​
Roles

Role: TypeAlias = Literal["agent", "user"]

​
Run Modes

RunMode: TypeAlias = Literal[
    "sync",    # Synchronous execution, wait for complete response
    "async",   # Asynchronous execution, don't wait for response
    "stream"   # Streaming execution, receive partial results
]

​
Trust Levels

TrustLevel: TypeAlias = Literal[
    "admin", "analyst", "auditor", "editor", "guest",
    "manager", "operator", "super_admin", "support", "viewer"
]

​
Error Codes

ErrorCode: TypeAlias = Literal["server_error", "invalid_input", "not_found"]

​
Negotiation Status

NegotiationStatus: TypeAlias = Literal["proposed", "accepted", "rejected", "countered"]
NegotiationSessionStatus: TypeAlias = Literal["initiated", "ongoing", "completed", "rejected"]

​
Identity Provider

IdentityProvider: TypeAlias = Literal["keycloak", "azure_ad", "okta", "auth0", "custom"]

​
Security & Authentication

The protocol supports multiple security schemes for API authentication:

​
HTTP Security Scheme

class HttpSecurityScheme(TypedDict):
    type: Required[Literal["http"]]
    scheme: Required[str]
    bearer_format: NotRequired[str]
    description: NotRequired[str]

​
API Key Security Scheme

class ApiKeySecurityScheme(TypedDict):
    type: Required[Literal["apiKey"]]
    name: Required[str]
    in_: Required[Literal["query", "header", "cookie"]]
    description: NotRequired[str]

​
OAuth2 Security Scheme

class OAuth2SecurityScheme(TypedDict):
    type: Required[Literal["oauth2"]]
    flows: Required[dict[str, Any]]
    description: NotRequired[str]

​
Agent Execution & Credits

The protocol includes types for managing agent execution costs and credit-based billing:

​
Agent Execution Cost

class AgentExecutionCost(TypedDict):
    agent_id: Required[str]
    agent_name: Required[str]
    credits_per_request: Required[int]
    creator_did: Required[str]
    minimum_trust_level: Required[TrustLevel]

​
Execution Request/Response

class ExecutionRequest(TypedDict):
    request_id: Required[UUID]
    executor_did: Required[str]
    agent_id: Required[str]
    input_data: Required[str]
    estimated_credits: Required[int]
    trust_level: Required[TrustLevel]

class ExecutionResponse(TypedDict):
    request_id: Required[UUID]
    execution_id: Required[UUID]
    success: Required[bool]
    credits_charged: Required[int]
    transaction_id: NotRequired[UUID]
    output_data: NotRequired[str]
    error_message: NotRequired[str]
    execution_time: Required[str]

​
Negotiation System

The protocol supports agent-to-agent negotiations:

​
Negotiation Proposal

class NegotiationProposal(TypedDict):
    proposal_id: Required[UUID]
    from_agent: Required[UUID]
    to_agent: Required[UUID]
    terms: Required[Dict[str, Any]]
    timestamp: Required[str]
    status: Required[NegotiationStatus]

​
Negotiation Session

class NegotiationSession(TypedDict):
    session_id: Required[UUID]
    status: Required[NegotiationSessionStatus]
    participants: Required[List[UUID]]
    proposals: Required[List[NegotiationProposal]]

​
Push Notifications

Support for real-time notifications outside of connected sessions: 
class PushNotificationConfig(TypedDict):
    id: Required[UUID]
    url: Required[str]
    token: NotRequired[str]
    authentication: NotRequired[SecurityScheme]

​
Event Types

The protocol defines events for real-time updates:

​
Task Status Update Event

class TaskStatusUpdateEvent(TypedDict):
    context_id: Required[UUID]
    final: Required[bool]
    kind: Required[Literal["status-update"]]
    metadata: NotRequired[dict[str, Any]]
    status: Required[TaskStatus]
    task_id: Required[UUID]

​
Task Artifact Update Event

class TaskArtifactUpdateEvent(TypedDict):
    append: NotRequired[bool]
    artifact: Required[Artifact]
    context_id: Required[UUID]
    kind: Required[Literal["artifact-update"]]
    last_chunk: NotRequired[bool]
    metadata: NotRequired[dict[str, Any]]
    task_id: Required[UUID]

​
JSON-RPC Protocol

The protocol uses JSON-RPC 2.0 for communication:

​
Base JSON-RPC Types

class JSONRPCMessage(TypedDict):
    jsonrpc: Required[Literal["2.0"]]
    id: Required[UUID]

class JSONRPCRequest(JSONRPCMessage, Generic[Method, Params]):
    method: Required[Method]
    params: Required[Params]

class JSONRPCResponse(JSONRPCMessage, Generic[ResultT, ErrorT]):
    result: NotRequired[ResultT]
    error: NotRequired[ErrorT]

class JSONRPCError(TypedDict, Generic[CodeT, MessageT]):
    code: Required[CodeT]
    message: Required[MessageT]
    data: NotRequired[Any]

​
Predefined Error Types

  • JSONParseError (-32700): Failed to parse JSON payload
  • InvalidRequestError (-32600): Request validation failed
  • MethodNotFoundError (-32601): Method not available
  • InvalidParamsError (-32602): Invalid parameters
  • InternalError (-32603): Internal server error
  • TaskNotFoundError (-32001): Task ID not found
  • TaskNotCancelableError (-32002): Task cannot be canceled
  • ContextNotFoundError (-32003): Context ID not found
  • ContextNotCancelableError (-32004): Context cannot be canceled
  • PushNotificationNotSupportedError (-32005): Push notifications not supported
  • UnsupportedOperationError (-32006): Operation not supported
  • ContentTypeNotSupportedError (-32007): Content type not supported
  • InvalidAgentResponseError (-32008): Agent response invalid

​
Request/Response Types

# Message operations
SendMessageRequest = JSONRPCRequest[Literal["message/send"], MessageSendParams]
StreamMessageRequest = JSONRPCRequest[Literal["message/stream"], MessageSendParams]

# Task operations
GetTaskRequest = JSONRPCRequest[Literal["tasks/get"], TaskQueryParams]
CancelTaskRequest = JSONRPCRequest[Literal["tasks/cancel"], TaskIdParams]
ListTasksRequest = JSONRPCRequest[Literal["tasks/list"], ListTasksParams]
TaskFeedbackRequest = JSONRPCRequest[Literal["tasks/feedback"], TaskFeedbackParams]

# Context operations
ListContextsRequest = JSONRPCRequest[Literal["contexts/list"], ListContextsParams]
ClearContextsRequest = JSONRPCRequest[Literal["contexts/clear"], ContextIdParams]

# Push notification operations
SetTaskPushNotificationRequest = JSONRPCRequest[Literal["tasks/pushNotification/set"], TaskPushNotificationConfig]
GetTaskPushNotificationRequest = JSONRPCRequest[Literal["tasks/pushNotification/get"], TaskIdParams]

​
Agent System

​
Agent Identity

class AgentIdentity(TypedDict):
    did: Required[str]  # Decentralized Identifier
    did_document: Required[Dict[str, Any]]
    agentdns_url: NotRequired[str]
    endpoint: NotRequired[str]
    public_key: Required[str]
    csr: Required[str]  # Certificate Signing Request

​
Agent Trust

class KeycloakRole(TypedDict):
    role_id: Required[UUID]
    role_name: Required[str]
    permissions: Required[List[str]]
    trust_level: Required[TrustLevel]
    realm_name: Required[str]
    external_mappings: Optional[Dict[str, str]]
    operation_permissions: Optional[Dict[str, TrustLevel]]

class AgentTrust(TypedDict):
    identity_provider: Required[IdentityProvider]
    inherited_roles: Required[List[KeycloakRole]]
    certificate: Optional[str]
    certificate_fingerprint: Optional[str]
    creator_id: Union[UUID, int, str]
    creation_timestamp: int
    trust_verification_required: bool
    allowed_operations: Dict[str, TrustLevel]

​
Agent Capabilities

class AgentSkill(TypedDict):
    id: Required[UUID]
    description: NotRequired[str]
    examples: NotRequired[list[str]]
    name: Required[str]
    tags: NotRequired[list[str]]

class AgentExtension(TypedDict):
    description: NotRequired[str]
    params: NotRequired[dict[str, Any]]
    required: Required[bool]
    uri: Required[str]

class AgentCapabilities(TypedDict):
    extensions: NotRequired[list[AgentExtension]]
    push_notifications: NotRequired[bool]
    state_transition_history: NotRequired[bool]
    streaming: NotRequired[bool]

​
Agent Card

class AgentCard(TypedDict):
    id: Required[UUID]
    name: Required[str]
    description: Required[str]
    url: Required[str]
    version: Required[str]
    protocol_version: Required[str]
    documentation_url: NotRequired[str]
    
    identity: Required[AgentIdentity]
    agent_trust: Required[AgentTrust]
    
    capabilities: Required[AgentCapabilities]
    skills: Required[List[AgentSkill]]
    
    kind: Required[Literal["agent", "team", "workflow"]]
    
    execution_cost: NotRequired[AgentExecutionCost]
    
    num_history_sessions: Required[int]
    extra_data: Required[Dict[str, Any]]
    
    debug_mode: Required[bool]
    debug_level: Required[Literal[1, 2]]
    
    monitoring: Required[bool]
    telemetry: Required[bool]

​
Parameter Types

​
Message Configuration

class MessageSendConfiguration(TypedDict):
    accepted_output_modes: Required[list[str]]
    blocking: NotRequired[bool]
    history_length: NotRequired[int]
    push_notification_config: NotRequired[PushNotificationConfig]

class MessageSendParams(TypedDict):
    configuration: Required[MessageSendConfiguration]
    message: Required[Message]
    metadata: NotRequired[dict[str, Any]]

​
Task Parameters

class TaskSendParams(TypedDict):
    task_id: Required[UUID]
    context_id: Required[UUID]
    message: NotRequired[Message]
    history_length: NotRequired[int]
    metadata: NotRequired[dict[str, Any]]

class TaskIdParams(TypedDict):
    task_id: Required[UUID]
    metadata: NotRequired[dict[str, Any]]

class TaskQueryParams(TaskIdParams):
    length: NotRequired[int]

class ListTasksParams(TypedDict):
    length: NotRequired[int]
    metadata: NotRequired[dict[str, Any]]

class TaskFeedbackParams(TypedDict):
    task_id: Required[UUID]
    feedback: Required[str]
    rating: NotRequired[int]  # Rating 1-5
    metadata: NotRequired[dict[str, Any]]

​
Context Parameters

class ContextIdParams(TypedDict):
    context_id: Required[UUID]
    metadata: NotRequired[dict[str, Any]]

class ContextQueryParams(ContextIdParams):
    history_length: NotRequired[int]

class ListContextsParams(TypedDict):
    length: NotRequired[int]
    metadata: NotRequired[dict[str, Any]]

​
Push Notification Parameters

class TaskPushNotificationConfig(TypedDict):
    id: Required[UUID]
    push_notification_config: Required[PushNotificationConfig]

class ListTaskPushNotificationConfigParams(TypedDict):
    id: Required[UUID]
    metadata: NotRequired[dict[str, Any]]

class DeleteTaskPushNotificationConfigParams(TypedDict):
    id: Required[UUID]
    push_notification_config_id: Required[UUID]
    metadata: NotRequired[dict[str, Any]]

​
Union Types

​
Discriminated Unions

# All message parts
Part = Annotated[Union[TextPart, FilePart, DataPart], Field(discriminator="kind")]

# All security schemes
SecurityScheme = Annotated[
    Union[HttpSecurityScheme, ApiKeySecurityScheme, OAuth2SecurityScheme, OpenIdConnectSecurityScheme],
    Discriminator("type")
]

# All request types
PebblingRequest = Annotated[
    Union[
        SendMessageRequest, StreamMessageRequest, GetTaskRequest, CancelTaskRequest,
        ListTasksRequest, TaskFeedbackRequest, ListContextsRequest, ClearContextsRequest,
        SetTaskPushNotificationRequest, GetTaskPushNotificationRequest, ResubscribeTaskRequest,
        ListTaskPushNotificationConfigRequest, DeleteTaskPushNotificationConfigRequest
    ],
    Discriminator("method")
]

# All response types
PebblingResponse: TypeAlias = Union[
    SendMessageResponse, StreamMessageResponse, GetTaskResponse, CancelTaskResponse,
    ListTasksResponse, TaskFeedbackResponse, ListContextsResponse, ClearContextsResponse,
    SetTaskPushNotificationResponse, GetTaskPushNotificationResponse, ResubscribeTaskResponse,
    ListTaskPushNotificationConfigResponse, DeleteTaskPushNotificationConfigResponse
]

​
Implementation Notes

  • All types use Pydantic for validation and serialization
  • CamelCase conversion is handled automatically via alias_generator
  • UUID types ensure unique identification across the system
  • TypedDict provides static type checking while maintaining runtime flexibility
  • Discriminated unions enable type-safe polymorphism for different message/artifact types
  • Generic types allow for flexible request/response patterns
  • TypeAdapters are provided for runtime validation and serialization
The protocol is designed to be extensible, type-safe, and backwards-compatible, ensuring robust communication between all system components.