Skip to main content

Models

This section provides detailed API documentation for the core data models in Protolink. These models represent the fundamental data structures used throughout the framework for agent communication, task management, and data exchange.

Protocol model layerCore Data Models

The stable dataclass and protocol vocabulary shared by agents, clients, servers, transports, registries, LLM wrappers, and storage-aware runtime features.

protolink.models
A2A-compatible cardsTask lifecycle stateMessages and partsArtifacts and endpointsLLM history contracts
IdentityAgent cards, capabilities, skills, roles, tags, security schemes, and advertised IO formats.AgentCard
Work unitsTask state, messages, parts, artifacts, errors, metadata, and helper constructors for inference.Task
RoutesEndpoint specifications let servers declare paths once and transports bind them to a backend.EndpointSpec
ContextLLM messages, conversation history, compaction requests, and compaction reports.ConversationHistory

Table of Contents​

AgentCard​

@dataclass
class AgentCard:
    name: str
    description: str
    url: str
    transport: TransportType = "http"
    version: str = "1.0.0"
    protocol_version: str = protolink_version
    capabilities: AgentCapabilities = field(default_factory=AgentCapabilities)
    skills: list[AgentSkill] = field(default_factory=list)
    input_formats: list[MimeType] = field(default_factory=lambda: ["text/plain"])
    output_formats: list[MimeType] = field(default_factory=lambda: ["text/plain"])
    security_schemes: dict[str, dict[str, Any]] | None = field(default_factory=dict)
    role: AgentRoleType = "worker"
    tags: list[str] = field(default_factory=list)

Agent identity and capability declaration. This is the main metadata card that describes an agent's identity, capabilities, and security requirements.

Parameters​

ParameterTypeDefaultDescription
namestr-Required. Agent name
descriptionstr-Required. Agent purpose/description
urlstr-Required. Service endpoint URL
transportTransportType"http"Transport protocol to use (e.g. "http", "sse", "websocket", "runtime")
versionstr"1.0.0"Agent version
protocol_versionstrprotolink_versionProtolink Protocol version
capabilitiesAgentCapabilitiesAgentCapabilities()Supported features
skillslist[AgentSkill][]List of skills the agent can perform
input_formatslist[MimeType]["text/plain"]Supported input MIME types
output_formatslist[MimeType]["text/plain"]Supported output MIME types
security_schemes`dict[SecuritySchemeType, dict[str, Any]]None`{}
roleAgentRoleType"worker"Agent role is a protocol-level contract that defines the agent's responsibility in the system topology (Extends A2A spec)
tagslist[str][]List of tags for categorization. These tags can be used for filtering during discovery (Protolink extension to A2A spec) E.g. "finance", "travel", "math" etc. (Extends A2A spec)

Methods​

to_dict() -> dict[str, Any]​

Convert the AgentCard to Python Dictionary / JSON format compatible with the A2A agent card specification.

Returns:

dict[str, Any]  # JSON dictionary representation

Example:

card = AgentCard(
    name="weather_agent",
    description="Weather service",
    url="https://api.example.com/weather",
)
json_data = card.to_dict()
print(json_data["name"])  # "weather_agent"

from_dict(data: dict[str, Any]) -> AgentCard classmethod​

Create an AgentCard from Python dict/JSON data. This method can also handle regular Python dictionaries and includes basic field validation via _validate_fields.

data: dict[str, Any]  # JSON dictionary or Python dict containing agent card data

Returns:

AgentCard  # New AgentCard instance

Example:

json_data = {
    "name": "weather_agent",
    "description": "Weather service",
    "url": "https://api.example.com/weather"
}
card = AgentCard.from_dict(json_data)

_validate_fields(data: dict[str, Any]) -> None staticmethod​

Validate the fields of the AgentCard.

Returns:

None

get_prompt_format() -> str​

Generate a structured, human-readable text representation of the agent and its skills for use in LLM system prompts. This is used by other agents to understand how to delegate tasks to this agent.

Returns:

str  # Multi-line formatted string

Example​

from protolink.models import AgentCard, AgentCapabilities

card = AgentCard(
    name="weather_agent",
    description="Provides weather information",
    url="https://api.example.com/weather",
    version="1.2.0",
    input_formats=["text/plain", "application/json"],
    output_formats=["text/plain", "application/json", "text/markdown"],
    capabilities=AgentCapabilities(
        streaming=True,
        tool_calling=True,
        max_concurrency=5
    )
)

# Convert to Python dict
json_data = card.to_dict()

AgentCapabilities​

@dataclass
class AgentCapabilities:
    streaming: bool = False
    push_notifications: bool = False
    state_transition_history: bool = False
    has_llm: bool = False
    max_concurrency: int = 1
    message_batching: bool = False
    tool_calling: bool = False
    multi_step_reasoning: bool = False
    timeout_support: bool = False
    delegation: bool = True
    rag: bool = False
    code_execution: bool = False

Defines the capabilities and limitations of an agent. This extends the A2A specification with additional capability flags.

Parameters​

ParameterTypeDefaultDescription
streamingboolFalseSupports streaming task events over a streaming-capable transport
push_notificationsboolFalseSupports push notifications (webhooks)
state_transition_historyboolFalseProvides detailed task state history
has_llmboolFalseHas an LLM component for AI processing
max_concurrencyint1Maximum concurrent tasks
message_batchingboolFalseProcesses multiple messages per request
tool_callingboolFalseCan call external tools/APIs
multi_step_reasoningboolFalsePerforms multi-step reasoning
timeout_supportboolFalseRespects operation timeouts
delegationboolTrueCan delegate tasks to other agents
ragboolFalseSupports Retrieval-Augmented Generation
code_executionboolFalseHas access to safe execution sandbox

Methods​

as_dict() -> dict[str, Any]​

Return all capabilities as a flat dictionary.

enabled() -> list[str]​

Return a list of all enabled (truthy) capabilities. Useful for quick inspection.

AgentSkill​

@dataclass
class AgentSkill:
    id: str
    description: str = ""
    input_schema: dict[str, Any] = field(default_factory=dict)
    output_schema: dict[str, Any] = field(default_factory=dict)
    tags: list[str] = field(default_factory=list)
    examples: list[str] = field(default_factory=list)

Represents a task that an agent can perform. Skills are used to advertise specific capabilities to other agents.

Parameters​

ParameterTypeDefaultDescription
idstr-Required. Unique Human-readable identifier for the task
descriptionstr""Detailed description of what the task does
input_schemadict[str, Any]{}JSON Schema object for accepted inputs
output_schemadict[str, Any]{}JSON Schema object for the output payload
tagslist[str][]List of tags for categorization
exampleslist[Any][]Example inputs, outputs, or usage scenarios

Example​

skill = AgentSkill(
    id="weather_forecast",
    description="Get weather forecast for any location",
    input_schema={
        "type": "object",
        "properties": {"location": {"type": "string"}},
        "required": ["location"],
        "additionalProperties": False,
    },
    output_schema={"type": "object", "additionalProperties": True},
    tags=["weather", "forecast", "location"],
    examples=[
        {"location": "New York"},
        {"location": "London"}
    ]
)

Type Aliases in AgentCard​

MimeType​

Type alias for supported MIME types in Protolink. These are used to specify the input and output formats that agents can handle.

Supported Types​

CategoryMIME Types
Texttext/plain, text/markdown, text/html
Structured Dataapplication/json
Imagesimage/png, image/jpeg, image/webp
Audioaudio/wav, audio/mpeg, audio/ogg
Videovideo/mp4, video/webm
Documentsapplication/pdf

SecuritySchemeType​

Type alias for supported security schemes in Protolink. These are used to specify the authentication methods that agents can use.

CategorySecurity Schemes
API keyapiKey
HTTP (bearer/basic/digest)http
OAuth 2.0oauth2
CertificatesmutualTLS
OIDC auto-discoveryopenIdConnect

Task​

@dataclass
class Task:
    id: str = field(default_factory=IDGenerator.generate_task_id)
    state: TaskState = TaskState.SUBMITTED
    messages: list[Message] = field(default_factory=list)
    artifacts: list[Artifact] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)
    flow_state: dict[str, Any] = field(default_factory=dict)
    created_at: str = field(default_factory=utc_now)

Unit of work exchanged between agents. Tasks encapsulate a complete unit of work including messages, state, and output artifacts.

Performance Note

The Task model implements internal caching for the most recent activity. Calling get_last_item() is an $O(1)$ operation, avoiding expensive list indexing or timestamp comparisons.

Parameters​

ParameterTypeDefaultDescription
idstruuid4()Unique task identifier
stateTaskStateSUBMITTEDCurrent task state
messageslist[Message][]Communication history
artifactslist[Artifact][]Output artifacts
metadatadict[str, Any]{}Additional metadata
flow_statedict[str, Any]{}Flow/orchestration context
created_atstrutc nowCreation timestamp

Lifecycle​

Task.state is an enforced lifecycle value, not a loose label. Valid transitions are:

submitted -> working -> completed
submitted -> working -> input-required -> working -> completed
submitted -> working -> failed
submitted -> failed
submitted -> canceled
input-required -> failed
input-required -> canceled

completed, failed, and canceled are terminal states. Once a task reaches one of them, it should not be processed again.

Every successful state change is recorded in task.metadata["state_history"]:

[
    {
        "previous_state": "submitted",
        "new_state": "working",
        "timestamp": "2026-06-12T08:30:00Z",
    }
]

The default Agent.execute_task() lifecycle is:

  1. Move a non-terminal task to WORKING.
  2. Execute explicit tool_call and infer parts from the latest message or artifact.
  3. Append outputs as artifacts or messages.
  4. Set the final state:
    • COMPLETED for successful outputs
    • FAILED for error parts, failed tool outputs, or exceptions
    • INPUT_REQUIRED for status parts requesting more input

Methods​

add_message(message: Message) -> Task​

Add a message to the task's communication history.

Parameters:

message: Message  # Message object to add to the task

Returns:

Task  # Self for method chaining

Example:

task = Task()
task.add_message(Message.user("What's the weather?"))
task.add_message(Message.agent("It's sunny!"))

add_artifact(artifact: Artifact) -> Task​

Add an output artifact to the task (v0.2.0+).

Parameters:

artifact: Artifact  # Artifact representing task output

Returns:

Task  # Self for method chaining

Example:

artifact = Artifact()
artifact.add_text("Weather analysis complete")
task.add_artifact(artifact)

update_state(state: TaskState | str) -> Task​

Transition the task to a new state.

The transition must be valid according to the lifecycle graph above. Repeating the current state is a no-op. Invalid transitions raise ValueError.

Parameters:

state: TaskState | str  # New task state

Returns:

Task  # Self for method chaining

Example:

task.update_state(TaskState.WORKING)
# ... process task ...
task.update_state(TaskState.COMPLETED)

Invalid direct jumps are rejected:

task = Task.create(Message.user("hello"))
task.update_state(TaskState.COMPLETED)
# ValueError: Invalid task state transition: submitted -> completed

begin() -> Task​

Mark the task as actively being processed.

This is equivalent to task.update_state(TaskState.WORKING).

require_input(message: Message | None = None) -> Task​

Mark the task as waiting for more input.

If the task is still SUBMITTED, this convenience method first transitions through WORKING so the lifecycle history remains valid.

complete(response_text: str) -> Task​

Mark the task as completed with a response message.

Parameters:

response_text: str  # Final response message text

Returns:

Task  # Self for method chaining

Example:

task.complete("The weather is sunny and 75°F.")
print(task.state)  # TaskState.COMPLETED
print(task.metadata["state_history"][-1]["new_state"])  # "completed"

fail(error_message: str) -> Task​

Mark the task as failed with an error message.

Parameters:

error_message: str  # Description of the error

Returns:

Task  # Self for method chaining

Example:

task.fail("Weather API unavailable")
print(task.state)  # TaskState.FAILED
print(task.metadata["error"])  # "Weather API unavailable"

cancel(reason: str | None = None) -> Task​

Mark the task as canceled.

This method updates lifecycle state only. To interrupt work currently executing on an Agent, use await agent.cancel_task(task.id, reason=...) or await client.cancel_task(agent_url, task.id, reason=...). See Runtime cancellation.

Parameters:

reason: str | None  # Optional cancellation reason stored in metadata

to_dict() -> dict[str, Any]​

Convert the task to a dictionary for serialization.

Returns:

dict[str, Any]  # Dictionary representation of the task

Example:

task_dict = task.to_dict()
print(task_dict["id"])  # Task UUID
print(task_dict["state"])  # Current state as string

from_dict(data: dict[str, Any]) -> Task classmethod​

Create a task from a dictionary.

Parameters:

data: dict[str, Any]  # Dictionary containing task data

Returns:

Task  # New Task instance

Example:

task_dict = {"state": "working", "messages": [], "artifacts": []}
task = Task.from_dict(task_dict)
print(task.state)  # TaskState.WORKING

create(message: Message) -> Task classmethod​

Create a new task with an initial message.

Parameters:

message: Message  # Initial message for the task

Returns:

Task  # New Task instance with the message added

Example:

task = Task.create(Message.user("Analyze this data"))
print(len(task.messages))  # 1
print(task.state)  # TaskState.SUBMITTED

create_infer(...) -> Task classmethod​

Create a new task initialized with an infer message.

Parameters:

  • prompt: str | None
  • user: str | None
  • output_schema: dict[str, Any] | None
  • metadata: dict[str, Any] | None

Returns:

Task  # New Task instance

Example:

task = Task.create_infer(prompt="Calculate this")

Example​

from protolink.models import Task, Message

# Create task with initial message
task = Task.create(Message.user("What's the weather in New York?"))

# Process it through the default lifecycle
task.begin()
task.complete("It's 72°F and sunny in New York.")

# Or use convenience method
task = Task.create(Message.user("Hello")).complete("Hi there!")

TaskState​

class TaskState(Enum):
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input-required"
    COMPLETED = "completed"
    CANCELED = "canceled"
    FAILED = "failed"
    UNKNOWN = "unknown"

Enumeration of possible task states.

Terminal States

COMPLETED, CANCELED, and FAILED are terminal. They intentionally have no outgoing transitions.

Values​

ValueDescription
SUBMITTEDTask has been submitted to the agent
WORKINGAgent is actively processing the task
INPUT_REQUIREDAgent needs additional input from user
COMPLETEDTask has been successfully completed
CANCELEDTask was canceled by user or agent
FAILEDTask failed due to an error
UNKNOWNTask state is unknown

Message​

@dataclass
class Message:
    id: str = field(default_factory=IDGenerator.generate_message_id)
    role: MessageRoleType = "user"
    parts: list[Part] = field(default_factory=list)
    timestamp: str = field(default_factory=utc_now)

Single unit of communication between agents. Messages contain one or more parts and have a specific role.

Parameters​

ParameterTypeDefaultDescription
idstruuid4()Unique message identifier
roleMessageRoleType"user"Sender role
partslist[Part][]Message content parts
timestampstrnowCreation timestamp

Role Types​

  • "user": Message from a human user
  • "agent": Message from an agent
  • "assistant": LLM assistant response
  • "system": System-level message

Methods​

add_text(text: str) -> Message​

Add a text part to the message.

Parameters:

text: str  # Text content to add

Returns:

Message  # Self for method chaining

Example:

msg = Message(role="user")
msg.add_text("Hello, world!")
print(len(msg.parts))  # 1
print(msg.parts[0].content)  # "Hello, world!"

add_part(part: Part) -> Message​

Add a content part to the message.

Parameters:

part: Part  # Part object to add (text, image, file, etc.)

Returns:

Message  # Self for method chaining

Example:

msg = Message(role="agent")
msg.add_part(Part.text("Here's the analysis:"))
msg.add_part(Part.json({"result": "success"}))

to_dict() -> dict[str, Any]​

Convert the message to a dictionary for serialization.

Returns:

dict[str, Any]  # Dictionary representation of the message

Example:

msg = Message.user("Hello")
msg_dict = msg.to_dict()
print(msg_dict["role"])  # "user"
print(msg_dict["parts"][0]["content"])  # "Hello"

from_dict(data: dict[str, Any]) -> Message classmethod​

Create a message from a dictionary.

Parameters:

data: dict[str, Any]  # Dictionary containing message data

Returns:

Message  # New Message instance

Example:

msg_dict = {
    "id": "msg-123",
    "role": "user",
    "parts": [{"type": "text", "content": "Hello"}],
    "timestamp": "2023-01-01T00:00:00Z"
}
msg = Message.from_dict(msg_dict)

user(text: str) -> Message classmethod​

Create a user message with text content (convenience method).

Parameters:

text: str  # Message text content

Returns:

Message  # New Message instance with role "user"

Example:

msg = Message.user("What's the weather?")
print(msg.role)  # "user"
print(msg.parts[0].content)  # "What's the weather?"

agent(text: str) -> Message classmethod​

Create an agent message with text content (convenience method).

Parameters:

text: str  # Message text content

Returns:

Message  # New Message instance with role "agent"

Example:

msg = Message.agent("It's sunny and 75°F.")
print(msg.role)  # "agent"
print(msg.parts[0].content)  # "It's sunny and 75°F."

assistant(text: str) -> Message classmethod​

Create an assistant message with text content.

infer(...) -> Message classmethod​

Create a user message with an infer part.

Parameters:

  • prompt: str | None
  • user: str | None
  • output_schema: dict[str, Any] | None
  • metadata: dict[str, Any] | None

Returns:

Message  # New Message instance with infer part

Example​

from protolink.models import Message, Part

# Create messages using convenience methods
user_msg = Message.user("What's the weather?")
agent_msg = Message.agent("It's sunny and 75°F.")

# Create message with multiple parts
msg = Message(role="user")
msg.add_text("Here's an image:")
msg.add_part(Part(type="image", content=image_data))

Part​

@dataclass
class Part:
    type: PartType
    content: Any

Atomic content unit within a message. Parts represent individual pieces of content like text, images, or files.

Parameters​

ParameterTypeDescription
typePartTypeContent type (e.g., 'text', 'image', 'file')
contentAnyThe actual content data

Methods​

to_dict() -> dict[str, Any]​

Convert to dictionary.

Returns: Dictionary representation

from_dict(data: dict[str, Any]) -> Part​

Create from dictionary.

Parameters:

  • data: Dictionary data

Returns: New Part instance

text(content: str) -> Part (classmethod)​

Create a text part.

Parameters:

  • content: Text content

Returns: New Part instance with type "text"

tool_call(...) -> Part classmethod​

Create a tool_call part.

tool_output(...) -> Part classmethod​

Create a tool_output part.

infer(...) -> Part classmethod​

Create an infer part.

infer_output(...) -> Part classmethod​

Create an infer_output part.

json(content: dict) -> Part classmethod​

Create a json part.

error(...) -> Part classmethod​

Create an error part.

status(...) -> Part classmethod​

Create a status part.

route(...) -> Part classmethod​

Create a structured flow route decision part with a route key plus optional reason, confidence, and metadata. Routers prefer this serialized control part over text tags.

decision(...) -> Part classmethod​

Create a structured decision part with the same content shape as route(...).

Example​

from protolink.models import Part

# Create different types of parts
text_part = Part.text("Hello, world!")
tool_call = Part.tool_call(tool_name="get_weather", args={"location": "Athens"})
infer_part = Part.infer(prompt="Who are you?")
route_part = Part.route("quality", reason="ready for review")

Artifact​

@dataclass
class Artifact:
    id: str = field(default_factory=IDGenerator.generate_artifact_id)
    parts: list[Part] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)
    timestamp: str = field(default_factory=utc_now)
    kind: str = "result"
    name: str | None = None
    uri: str | None = None
    media_type: str | None = None
    action_id: str | None = None

Structured output or preview produced during a run. Artifacts represent task results, resources, diagnostics, action previews, or other durable output without imposing an application-specific taxonomy.

Parameters​

ParameterTypeDefaultDescription
idstruuid4()Unique artifact identifier
partslist[Part][]Content parts of the artifact
metadatadict[str, Any]{}Artifact metadata
timestampstrutc nowCreation timestamp
kindstr"result"Extensible artifact category
namestr ⎪ NoneNoneOptional display or resource name
uristr ⎪ NoneNoneOptional represented resource URI
media_typestr ⎪ NoneNoneOptional MIME type
action_idstr ⎪ NoneNoneRelated runtime action identifier

Methods​

add_part(part: Part) -> Artifact​

Add content part to artifact.

Parameters:

  • part: Part to add

Returns: Self for method chaining

add_text(text: str) -> Artifact​

Add text content (convenience method).

Parameters:

  • text: Text content

Returns: Self for method chaining

for_action(action_id: str) -> Artifact​

Associate the artifact with a RunAction and return it for method chaining.

to_dict() -> dict[str, Any]​

Convert to dictionary.

Returns: Dictionary representation

from_dict(data: dict[str, Any]) -> Artifact​

Create from dictionary.

Parameters:

  • data: Dictionary data

Returns: New Artifact instance

Example​

from protolink.models import Artifact, Part

# Create artifact with multiple parts
artifact = Artifact(kind="diagnostic", name="analysis report", media_type="application/json")
artifact.add_text("Analysis Results:")
artifact.add_part(Part.json({"results": [1, 2, 3]}))
artifact.add_part(Part(type="image", content=chart_image_data))

# Set metadata
artifact.metadata["type"] = "analysis_report"
artifact.metadata["version"] = "1.0"

EndpointSpec​

@dataclass(frozen=True)
class EndpointSpec:
    name: str
    path: str
    method: HttpMethod
    handler: Callable[..., Awaitable]
    content_type: Literal["json", "html"] = "json"
    streaming: bool = False
    mode: Literal["request_response", "stream"] = "request_response"
    request_parser: Callable[[Any], Any] | None = None
    request_source: RequestSourceType = "none"

Defines the contract for a server endpoint. This model lives in protolink.server.endpoint_handler and bridges the gap between the server implementation (Agent/Registry) and the underlying transport.

Parameters​

ParameterTypeDefaultDescription
namestr-Required. Unique internal name for the endpoint
pathstr-Required. URL path (e.g., /tasks/)
methodHttpMethod-Required. HTTP method (GET, POST, etc.)
handlerCallable-Required. Async function that handles the request
content_typestr"json"Response content type (json or html)
streamingboolFalseWhether the endpoint handler returns a stream of events
modestr"request_response"Interaction mode (request_response or stream)
request_parserCallableNoneOptional function to parse raw request data
request_sourceRequestSourceType"none"Source of request data (body, query_params, etc.)

LLMMessage​

@dataclass
class LLMMessage:
    role: LLMMessageRole
    content: str
    name: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    id: str = field(default_factory=lambda: str(uuid4()))
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    tool_calls: dict[str, Any] = field(default_factory=dict)
    tool_name: str | None = None

Canonical message format used internally across all LLM providers. Unlike the standard Message model used for agent-to-agent communication, LLMMessage is optimized for LLM context windows and provider-specific schemas.

Methods​

to_dict() -> dict[str, Any]​

Convert the message to a JSON-serializable dictionary, preserving all fields (including metadata and tool calls).

from_dict(data: dict[str, Any]) -> LLMMessage classmethod​

Create an LLMMessage instance from a dictionary.

ConversationHistory​

class ConversationHistory:
    def __init__(self, system_prompt: str | None = None):
        self._messages: deque[LLMMessage] = deque()

Manages conversation state in a provider-agnostic way.

Optimized Architecture

ConversationHistory uses a collections.deque to ensure that updating system prompts or truncating history remains highly performant ($O(1)$ prepends) even as the conversation grows.

Methods​

to_list() -> list[dict[str, Any]]​

Convert the entire history into a list of full message dictionaries. Used for persistence.

replace(messages_data: Iterable[dict[str, Any]]) -> None​

Replace all messages from full message dictionaries while preserving the ConversationHistory object's identity. Used by history transformations such as LLM.compact_history().

from_list(messages_data: list[dict[str, Any]]) -> ConversationHistory classmethod​

Restore a conversation history from a list of message dictionaries.

add_user(content: str)​

Add a user message to history.

add_assistant(content: str)​

Add an assistant message to history.

add_system(content: str)​

Add a system message to history.

reset_to_system(system_prompt: str)​

Wipes all history and sets a new system prompt.

set_system(system_prompt: str)​

Updates the system prompt (at index 0) while preserving all subsequent conversation turns.

truncate(max_messages: int)​

Keep at most max_messages messages using the low-level deque truncation operation. For provider-neutral compaction with structured results, token budgets, and summarization, prefer LLM.compact_history().

HistoryCompactionRequest​

@dataclass(frozen=True)
class HistoryCompactionRequest:
    strategy: Literal["recent", "tokens", "summary"] = "recent"
    max_messages: int = 20
    max_tokens: int = 4000
    preserve_recent: int = 6
    summary_max_tokens: int = 512
    session_id: str | None = None
    metadata: dict[str, Any] | None = None

Transport-neutral control payload used by Agent.compact_history() and AgentClient.compact_history(). It is not a Task part and is not shown to the model.

HistoryCompactionResult​

@dataclass(frozen=True)
class HistoryCompactionResult:
    strategy: Literal["recent", "tokens", "summary"]
    before_messages: int
    after_messages: int
    removed_messages: int
    before_tokens: int
    after_tokens: int
    summary_created: bool = False

Structured report returned by direct LLM compaction, Agent control-plane compaction, and the client request spec.