Skip to main content

Models

ProtoLink's models are the transport-neutral vocabulary shared by agents, clients, servers, flows, storage, telemetry, registries, and LLM adapters. They describe who an agent is, how work moves through its lifecycle, what a message contains, and how those values cross process boundaries.

These are ProtoLink's ergonomic runtime forms of A2A's core agent primitives. The A2A 1.0 adapter maps the advertised subset to canonical wire models when interoperability is required; the classes on this page remain the native Python contract used inside ProtoLink.

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-derived cards and tasksTask 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

Package overview​

Most application-facing models can be imported from either protolink or protolink.models:

from protolink import AgentCard, AgentInterface, AgentSkill
from protolink import Artifact, Message, Part, Task, TaskState
from protolink import HistoryCompactionRequest, HistoryCompactionResult
from protolink.models import EndpointSpec, RouteDecision

The source files are partitioned by runtime responsibility:

  • protolink.core.agent_card owns identity, capabilities, skills, and additional interfaces.
  • protolink.core.task, message, part, and artifact own the task envelope and its nested content.
  • protolink.server.endpoint_handler owns transport-neutral server endpoint declarations.
  • protolink.llms.history owns the provider-neutral LLM context representation.
  • protolink.llms.compaction owns the request and result values used by direct, agent, and client compaction APIs.
  • protolink.models is a convenience re-export layer. It intentionally gathers models from those focused modules rather than defining a second set of classes.

AgentCapabilities, LLMMessage, and ConversationHistory are lower-level implementation-facing types and are imported from their defining modules. They are documented here because they are important when customizing discovery cards or LLM state:

from protolink.core.agent_card import AgentCapabilities
from protolink.llms.history import ConversationHistory, LLMMessage, LLMMessageRole
Serialization boundaries

to_dict() methods return ordinary Python dictionaries suitable for ProtoLink's JSON boundaries, but they are not universal schema validators and generally do not deep-copy arbitrary metadata. Constructing a dataclass directly is intentionally permissive in several places; from_dict() performs the normalization required by that class's wire format.

Table of Contents​

Messages and content​

Messages are the ordered communication units inside a task. Each message contains one or more Part values so plain text, structured control requests, tool calls, routes, errors, and media can share the same envelope. Artifacts use the same part model for durable outputs.

Message​

dataclassprotolink.Message
source
class Message(
  id: str = generate_message_id(),
  role: MessageRoleType = "user",
  parts: list[Part] = [],
  timestamp: str = utc_now(),
)

One unit of communication between a user, agent, assistant model, or system layer. The message envelope supplies identity, sender role, and creation time; ordered Part values carry the actual text, structured data, control request, or result.

Parameters

idstrdefault: generate_message_id()

Unique message identifier. The default generator creates a msg_-prefixed timestamp plus a random suffix.

roleMessageRoleTypedefault: "user"

Sender role: user, agent, assistant, or system. The literal annotation is not enforced at direct dataclass construction, so use the convenience constructors when possible.

partslist[Part]default: []

Ordered message content. A fresh list is created for each default instance; a list supplied by the caller is stored directly.

timestampstrdefault: utc_now()

ISO 8601 UTC creation time used when comparing messages with artifacts for a task's last-item cache.

Message versus LLMMessage

Message is the agent/task protocol envelope and may contain heterogeneous parts. LLMMessage is the provider-neutral context entry used by LLM adapters and always carries a text content field.

Message.add_text​

methodprotolink.Message.add_text
source
add_text(
  text: str,
) -> Message

Append plain text by constructing Part.text(text).

Parameters

textstrrequired

Text content to append. The method stores the string as-is and does not trim, normalize, or reject an empty value.

Returns

selfMessage

The mutated message for method chaining.

Side effect

Appends one new part to parts. The message identifier, role, and timestamp remain unchanged.

Message.add_part​

methodprotolink.Message.add_part
source
add_part(
  part: Part,
) -> Message

Append an existing part to the message. Use this for structured JSON, media, tool calls, inference requests, route decisions, and custom part types.

Parameters

partPartrequired

Content part to append. No runtime type check or defensive copy is performed.

Returns

selfMessage

The mutated message.

Message.to_dict​

methodprotolink.Message.to_dict
source
to_dict() -> dict[str, Any]

Serialize the message and each nested part into the native task wire shape.

Returns

datadict[str, Any]

Dictionary containing id, role, serialized parts, and timestamp.

Content conversion

Each part controls conversion of its own content. Dataclass-backed tool and route content becomes a dictionary; arbitrary custom content is returned as supplied and must already be compatible with the eventual encoder.

Message.from_dict​

classmethodprotolink.Message.from_dict
source
from_dict(
  data: dict[str, Any],
) -> Message

Create a message from native serialized data and hydrate every nested part.

Parameters

datadict[str, Any]required

Message mapping. Missing identifier, role, parts, or timestamp receive the same defaults as direct construction.

Returns

messageMessage

A new message whose nested tool calls, tool outputs, and route decisions are normalized by Part.from_dict().

Raises

KeyError, TypeError, or ValueError

Propagated from malformed nested part payloads. The message mapping itself has no mandatory keys because defaults are available for every constructor field.

Examples

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

Message.user​

classmethodprotolink.Message.user
source
user(
  text: str,
) -> Message

Create a user-role message containing one text part. This is the normal constructor for human or calling-client input.

Parameters

textstrrequired

User text stored in a new Part(type="text", ...).

Returns

messageMessage

New message with role user, a generated identifier and timestamp, and one text part.

Message.agent​

classmethodprotolink.Message.agent
source
agent(
  text: str,
) -> Message

Create an agent-role message containing one text part. Task convenience methods such as Task.complete() use this constructor for final responses.

Parameters

textstrrequired

Agent response text.

Returns

messageMessage

New message with role agent and one text part.

Message.assistant​

classmethodprotolink.Message.assistant
source
assistant(
  text: str,
) -> Message

Create an assistant-role message containing one text part. Use this role when preserving the distinction between an LLM assistant response and the broader agent runtime identity.

Parameters

textstrrequired

Assistant response text.

Returns

messageMessage

New message with role assistant and one text part.

Message.route​

classmethodprotolink.Message.route
source
route(
  route_key: str,
  *,
  reason: str | None = None,
  confidence: float | None = None,
  metadata: dict[str, Any] | None = None,
) -> Message

Create an agent-role message containing one structured route decision. Routers can inspect the typed decision instead of parsing fragile text labels.

Parameters

route_keystrrequired

Key expected by the receiving router's route map.

reasonstr | Nonedefault: None

Optional human-readable explanation for observability or debugging.

confidencefloat | Nonedefault: None

Optional confidence score. The model does not clamp or validate the documented zero-to-one range.

metadatadict[str, Any] | Nonedefault: None

Additional serializable decision context. None and an empty dictionary both become a fresh empty mapping.

Returns

messageMessage

New agent message containing Part.route(...).

Message.infer​

classmethodprotolink.Message.infer
source
infer(
  *,
  prompt: str | None = None,
  user: str | None = None,
  output_schema: dict[str, Any] | None = None,
  metadata: dict[str, Any] | None = None,
) -> Message

Create a user-role message containing one infer control part. When an agent executes the enclosing task, the part requests an LLM inference rather than representing ordinary display text.

Parameters

promptstr | Nonedefault: None

Main model instruction included when supplied.

userstr | Nonedefault: None

Optional user context carried inside the control payload.

output_schemadict[str, Any] | Nonedefault: None

Optional schema for a structured response.

metadatadict[str, Any] | Nonedefault: None

Optional operation metadata, distinct from message and task metadata.

Returns

messageMessage

New user message containing exactly one infer part. Values that are None are omitted from the part content.

Message.tool_call​

classmethodprotolink.Message.tool_call
source
tool_call(
  *,
  tool_name: str,
  args: dict[str, Any] | None = None,
  call_id: str | None = None,
) -> Message

Create a user-role message containing one typed tool invocation. The receiving agent resolves and executes the tool when processing the task.

Parameters

tool_namestrrequired

Canonical registered tool name.

argsdict[str, Any] | Nonedefault: None

Tool arguments. None becomes an empty mapping.

call_idstr | Nonedefault: None

Optional correlation identifier used to match a later tool_output. A generated identifier is used when omitted.

Returns

messageMessage

New user message containing exactly one tool-call part.

Message examples

from protolink import Message, Part

user_message = Message.user("What's the weather?")
agent_message = Message.agent("It's sunny and 24°C.")

multi_part = (
    Message(role="user")
    .add_text("Analyze this payload:")
    .add_part(Part.json({"city": "Athens"}))
)

route_message = Message.route(
    "quality",
    reason="Draft is ready for review",
    confidence=0.92,
)

Part​

dataclassprotolink.Part
source
class Part(
  type: PartType,
  content: Any,
)

Atomic content unit within a message or artifact. The type is the dispatch key; content may be text, JSON-compatible data, media, a typed tool-call value, a typed tool output, or a route decision.

Parameters

typePartTyperequired

Content category such as text, json, tool_call, tool_output, infer, route, error, or a supported media type. Direct construction does not runtime-check the literal.

contentAnyrequired

Payload interpreted according to type. Direct construction preserves it unchanged; from_dict() hydrates selected structured part types into dataclasses.

Prefer factories for structured parts

Use tool_call(), tool_output(), route(), and decision() to obtain typed content with generated IDs and normalized metadata. A direct Part(type="tool_call", content=dict(...)) remains dictionary-backed until explicitly converted.

Part.to_dict​

methodprotolink.Part.to_dict
source
to_dict() -> dict[str, Any]

Serialize a part into its two-field native representation.

Returns

datadict[str, Any]

Dictionary containing type and content. Dataclass content, including tool calls, tool outputs, and route decisions, is recursively converted with dataclasses.asdict(); other content is returned unchanged.

JSON compatibility

The method does not encode bytes, arbitrary objects, or custom mappings. A part can be valid in memory while still requiring a transport-specific encoder.

Part.from_dict​

classmethodprotolink.Part.from_dict
source
from_dict(
  data: dict[str, Any],
) -> Part

Rehydrate a serialized part. Tool-call, tool-output, route, and decision dictionaries become their typed dataclass representations; all other content remains as supplied.

Parameters

datadict[str, Any]required

Mapping with a required type and optional content. Missing content becomes None.

Returns

partPart

New part with normalized structured content where supported.

Raises

KeyError

Raised when type is absent, or when a serialized tool call does not provide its required tool_name.

ValueError

Raised when route or decision content lacks route_key and its accepted compatibility aliases route and key.

Generated correlation IDs

A serialized tool call or output without call_id receives a newly generated identifier during hydration. Supply the original ID when correlation must survive a round trip.

Part.as_tool_call​

methodprotolink.Part.as_tool_call
source
as_tool_call() -> ToolCall

Validate that the part represents a tool call and return a typed ToolCall view of its content.

Returns

tool_callToolCall

Existing typed content, or a newly hydrated value when the content is a dictionary.

Raises

ValueError

Raised when type is not tool_call.

TypeError

Raised when the type is correct but content is neither ToolCall nor a dictionary.

KeyError

Raised when dictionary content lacks tool_name.

No in-place conversion

When content is dictionary-backed, this method returns a hydrated object but does not assign it back to part.content.

Part.as_tool_output​

methodprotolink.Part.as_tool_output
source
as_tool_output() -> ToolOutput

Validate that the part represents a tool result and return a typed ToolOutput view.

Returns

tool_outputToolOutput

Existing typed output or a newly hydrated view of dictionary content.

Raises

ValueError

Raised when type is not tool_output.

TypeError

Raised when content has an unsupported runtime type.

Part.as_route_decision​

methodprotolink.Part.as_route_decision
source
as_route_decision() -> RouteDecision

Read a typed route decision from either a route or decision part.

Returns

decisionRouteDecision

Existing typed decision or a hydrated view of dictionary content.

Raises

ValueError

Raised for any part type other than route or decision, or when dictionary content lacks a route key.

TypeError

Raised when content is neither a RouteDecision nor a dictionary.

Part.text​

classmethodprotolink.Part.text
source
text(
  content: str,
) -> Part

Create a plain-text part.

Parameters

contentstrrequired

Text payload preserved exactly as supplied.

Returns

partPart

New part with type text.

Part.json​

classmethodprotolink.Part.json
source
json(
  content: dict,
) -> Part

Create a structured JSON part. The method labels the mapping but does not serialize or copy it.

Parameters

contentdictrequired

Mapping to store as the part content. Nested values must be serializable by the eventual transport.

Returns

partPart

New part with type json.

Part.error​

classmethodprotolink.Part.error
source
error(
  code: str,
  message: str,
  *,
  retryable: bool = False,
) -> Part

Create a structured error part suitable for task failure detection and client display.

Parameters

codestrrequired

Stable machine-readable error identifier.

messagestrrequired

Human-readable failure explanation.

retryablebooldefault: False

Advisory flag indicating whether repeating the operation may succeed. It does not schedule a retry.

Returns

partPart

Error part whose content contains code, message, and retryable.

Part.status​

classmethodprotolink.Part.status
source
status(
  state: str,
  message: str | None = None,
) -> Part

Create a structured status part. Agent lifecycle handling can use status content to communicate progress or request additional input.

Parameters

statestrrequired

Application or runtime status label. This helper does not coerce the value to TaskState.

messagestr | Nonedefault: None

Optional human-readable status detail. The key remains present with a null value when omitted.

Returns

partPart

Status part containing state and message.

Part.route​

classmethodprotolink.Part.route
source
route(
  route_key: str,
  *,
  reason: str | None = None,
  confidence: float | None = None,
  metadata: dict[str, Any] | None = None,
) -> Part

Create a structured flow-routing part backed by RouteDecision. Routers prefer this typed control value over extracting route names from prose.

Parameters

route_keystrrequired

Destination key in the receiving router's route map.

reasonstr | Nonedefault: None

Optional explanation for the selection.

confidencefloat | Nonedefault: None

Optional confidence score. No numeric range validation is performed.

metadatadict[str, Any] | Nonedefault: None

Additional serializable context; falsy values become a fresh empty dictionary.

Returns

partPart

Part with type route and typed RouteDecision content.

Part.decision​

classmethodprotolink.Part.decision
source
decision(
  route_key: str,
  *,
  reason: str | None = None,
  confidence: float | None = None,
  metadata: dict[str, Any] | None = None,
) -> Part

Create a structured decision part with the same RouteDecision content as route(). Use the distinct type when an application wants to label a branching decision without calling it a route.

Parameters

route_keystrrequired

Selected branch or decision key.

reasonstr | Nonedefault: None

Optional human-readable rationale.

confidencefloat | Nonedefault: None

Optional unvalidated confidence value.

metadatadict[str, Any] | Nonedefault: None

Optional additional decision context.

Returns

partPart

Part with type decision and typed route-decision content.

Part.tool_call​

classmethodprotolink.Part.tool_call
source
tool_call(
  *,
  tool_name: str,
  args: dict[str, Any] | None = None,
  call_id: str | None = None,
) -> Part

Create a standardized tool or capability invocation. The typed content keeps the tool name, arguments, and correlation identifier together through task serialization.

Parameters

tool_namestrrequired

Canonical name resolved by the receiving agent's tool registry.

argsdict[str, Any] | Nonedefault: None

Arguments passed to the tool. None and other falsy mappings become a new empty dictionary.

call_idstr | Nonedefault: None

Correlation identifier used by the corresponding tool output. A generated tool_call_-prefixed ID is retained when this value is None.

Returns

partPart

Part with type tool_call and a typed ToolCall content object.

Resolution timing

Creating the part does not verify that the named tool exists or validate its arguments. Those checks happen when an agent executes the call.

Part.tool_output​

classmethodprotolink.Part.tool_output
source
tool_output(
  *,
  call_id: str | None = None,
  result: Any | None = None,
  error: dict | None = None,
) -> Part

Create the success or failure result for an earlier tool call.

Parameters

call_idstr | Nonedefault: None

Identifier of the originating call. If omitted, a new tool_output_-prefixed ID is generated; that generated value will not correlate with an earlier call unless the caller records it explicitly.

resultAny | Nonedefault: None

Successful result payload. ProtoLink does not enforce mutual exclusivity with error.

errordict | Nonedefault: None

Structured error payload for a failed invocation.

Returns

partPart

Part with type tool_output and typed ToolOutput content.

Part.infer​

classmethodprotolink.Part.infer
source
infer(
  *,
  prompt: str | None = None,
  user: str | None = None,
  output_schema: dict[str, Any] | None = None,
  metadata: dict[str, Any] | None = None,
) -> Part

Create a control part instructing an agent to invoke its configured LLM. This is the low-level value wrapped by Message.infer() and Task.create_infer().

Parameters

promptstr | Nonedefault: None

Model instruction.

userstr | Nonedefault: None

Optional user identity or context.

output_schemadict[str, Any] | Nonedefault: None

Optional structured-output schema.

metadatadict[str, Any] | Nonedefault: None

Optional operation metadata.

Returns

partPart

Part with type infer. Every argument whose value is None is removed from the content dictionary; empty strings and empty mappings remain.

Part.infer_output​

classmethodprotolink.Part.infer_output
source
infer_output(
  *,
  content: str | dict[str, Any],
) -> Part

Wrap the result of an LLM inference operation in a dedicated output part.

Parameters

contentstr | dict[str, Any]required

Unstructured response text or a structured result mapping. The value is stored without copying or schema validation.

Returns

partPart

Part with type infer_output.

Part examples

from protolink import Part

text_part = Part.text("Hello, world!")
json_part = Part.json({"status": "ready"})

tool_call = Part.tool_call(
    tool_name="get_weather",
    args={"location": "Athens"},
)
tool_result = Part.tool_output(
    call_id=tool_call.as_tool_call().call_id,
    result={"temperature": 24},
)

infer_part = Part.infer(prompt="Summarize the weather.")
route_part = Part.route("quality", reason="Ready for review")

RouteDecision​

dataclassprotolink.models.RouteDecision
source
class RouteDecision(
  route_key: str,
  reason: str | None = None,
  confidence: float | None = None,
  metadata: dict[str, Any] = {},
)

Typed content carried by route and decision parts. Keeping the selected key separate from explanatory text lets Router branch deterministically while retaining rationale, confidence, and application context for observability.

Parameters

route_keystrrequired

Selected key in the receiving router's route map. The model stores the value without confirming that a matching route exists.

reasonstr | Nonedefault: None

Optional human-readable rationale for logs, traces, or review.

confidencefloat | Nonedefault: None

Optional confidence score. Although zero to one is the intended semantic range, this dataclass does not enforce it.

metadatadict[str, Any]default: {}

Additional serializable routing context. Each default instance receives an independent mapping.

Normal construction

Applications usually call Part.route(), Part.decision(), or Message.route(). Those helpers wrap this value with the appropriate part and message type.

Artifact​

dataclassprotolink.Artifact
source
class Artifact(
  id: str = generate_artifact_id(),
  parts: list[Part] = [],
  metadata: dict[str, Any] = {},
  timestamp: str = 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 carry the same flexible parts as messages while adding durable descriptors for resources, diagnostics, previews, and action-related results.

Parameters

idstrdefault: generate_artifact_id()

Unique artifact identifier generated with an art_ prefix by default.

partslist[Part]default: []

Ordered output content, such as text, JSON, media, tool output, or an inference result.

metadatadict[str, Any]default: {}

Extensible application metadata. The model imposes no reserved schema.

timestampstrdefault: utc_now()

ISO 8601 UTC creation time used by task last-item ordering.

kindstrdefault: "result"

Application-defined category such as result, preview, or diagnostic. It remains a free string so domains can extend the taxonomy.

namestr | Nonedefault: None

Optional display name or represented resource name.

uristr | Nonedefault: None

Optional URI identifying the represented resource.

media_typestr | Nonedefault: None

Optional MIME type describing the artifact as a whole. Individual parts may still carry heterogeneous content.

action_idstr | Nonedefault: None

Optional identifier of the RunAction that produced or proposes this artifact.

Descriptor semantics

Optional descriptors are informational and are not cross-validated. For example, setting media_type does not transform parts or verify that their content matches the MIME type.

Artifact.add_part​

methodprotolink.Artifact.add_part
source
add_part(
  part: Part,
) -> Artifact

Append an existing content part to the artifact.

Parameters

partPartrequired

Part to append. It is stored by reference without validation or copying.

Returns

selfArtifact

The mutated artifact for chaining.

Artifact.add_text​

methodprotolink.Artifact.add_text
source
add_text(
  text: str,
) -> Artifact

Append a plain-text part to the artifact.

Parameters

textstrrequired

Text wrapped by Part.text().

Returns

selfArtifact

The mutated artifact.

Artifact.for_action​

methodprotolink.Artifact.for_action
source
for_action(
  action_id: str,
) -> Artifact

Associate the artifact with a runtime action. This is useful when a preview is created before the final RunAction identifier is known.

Parameters

action_idstrrequired

Identifier assigned directly to the artifact. It is not checked against an action registry.

Returns

selfArtifact

The same artifact after mutation.

Artifact.to_dict​

methodprotolink.Artifact.to_dict
source
to_dict() -> dict[str, Any]

Serialize all artifact fields and nested parts into the native dictionary representation.

Returns

datadict[str, Any]

Dictionary containing every descriptor key, including optional keys whose values are None.

Copy semantics

Parts are converted recursively. The metadata mapping and non-dataclass part contents are not deep-copied.

Artifact.from_dict​

classmethodprotolink.Artifact.from_dict
source
from_dict(
  data: dict[str, Any],
) -> Artifact

Create an artifact from serialized data while remaining compatible with payloads emitted before structured descriptor fields were added.

Parameters

datadict[str, Any]required

Artifact mapping. Missing identifiers and timestamps are generated, missing or falsy kind becomes result, and missing metadata becomes a fresh empty dictionary.

Returns

artifactArtifact

New artifact with hydrated parts. Non-None values for name, uri, media_type, and action_id are converted with str().

Raises

KeyError, TypeError, or ValueError

Propagated from malformed nested parts or from values that cannot be converted to the expected container shape.

Examples

from protolink import Artifact, Part

artifact = (
    Artifact(
        kind="diagnostic",
        name="analysis report",
        media_type="application/json",
    )
    .add_text("Analysis results:")
    .add_part(Part.json({"results": [1, 2, 3]}))
    .for_action("action_42")
)

artifact.metadata["version"] = "1.0"

Agent identity​

Discovery starts with an AgentCard. The card identifies one logical agent, describes the work it can perform, and advertises how peers can reach it. Capabilities are coarse feature flags; skills provide task-level schemas and examples; interfaces describe alternate endpoints for the same identity.

AgentCard​

dataclassprotolink.AgentCard
source
class AgentCard(
  name: str,
  description: str,
  url: str,
  transport: TransportType = "http",
  version: str = "1.0.0",
  protocol_version: str = protolink_version,
  capabilities: AgentCapabilities = AgentCapabilities(),
  skills: list[AgentSkill] = [],
  input_formats: list[MimeType] = ["text/plain"],
  output_formats: list[MimeType] = ["text/plain"],
  security_schemes: dict[SecuritySchemeType, dict[str, Any]] | None = {},
  role: AgentRoleType = "worker",
  tags: list[str] = [],
  interfaces: list[AgentInterface] = [],
)

Agent identity and capability declaration used by ProtoLink discovery, registration, delegation prompts, and server metadata. The primary url and transport describe the normal route; interfaces advertises additional routes to the same logical agent.

Parameters

namestrrequired

Stable human-readable identity used by registries, delegation prompts, logs, and agent lookup. from_dict() rejects an absent, empty, or otherwise falsy name, although direct dataclass construction does not repeat that validation.

descriptionstrrequired

Clear explanation of the agent's purpose and when another agent should delegate work to it. This text is included in get_prompt_format(), so operational descriptions are more useful than marketing copy.

urlstrrequired

Primary service endpoint. ProtoLink stores the value as supplied; URL syntax and reachability are validated later by the selected transport rather than by the dataclass.

transportTransportTypedefault: "http"

Registered transport for the primary URL. Supported annotations include http, websocket, sse, json-rpc, sse-json-rpc, grpc, and runtime. Runtime construction does not independently validate the literal.

versionstrdefault: "1.0.0"

Application-defined version of the agent implementation. It lets clients distinguish behavior changes independently from the protocol version.

protocol_versionstrdefault: protolink_version

ProtoLink native-card protocol version. The default is resolved from the installed package version when the module is imported. A2A adapters own their interface version separately.

capabilitiesAgentCapabilities | Mapping[str, Any]default: AgentCapabilities()

Coarse features and limits advertised by the agent. A mapping is normalized into AgentCapabilities during post_init; missing mapping keys receive dataclass defaults. Any other object raises TypeError.

skillslist[AgentSkill]default: []

Specific operations available for discovery and delegation. Each skill can carry input and output JSON Schemas plus examples. Unlike capabilities, raw skill mappings are not normalized by direct construction; use AgentSkill objects or AgentCard.from_dict().

input_formatslist[MimeType]default: ["text/plain"]

MIME types accepted by the agent's normal task interface. Each card receives an independent list from the default factory.

output_formatslist[MimeType]default: ["text/plain"]

MIME types the agent may return. This is discovery metadata, not automatic response transcoding.

security_schemesdict[SecuritySchemeType, dict[str, Any]] | Nonedefault: {}

Named authentication scheme declarations used by discovery consumers. ProtoLink preserves the mapping as supplied and does not validate the nested OpenAPI-style scheme definition here. None is accepted and serialized as null.

roleAgentRoleTypedefault: "worker"

Native runtime responsibility such as a worker or orchestrator. This field is available in memory, but the current native to_dict() and from_dict() paths do not serialize or restore it.

tagslist[str]default: []

Discovery labels such as finance, travel, or math. Tags are serialized verbatim and are suitable for registry-side filtering.

interfaceslist[AgentInterface | Mapping[str, Any]]default: []

Additional URLs and transports for the same identity. Mappings are normalized through AgentInterface.from_dict(). They serialize under additionalInterfaces, which is distinct from the A2A 1.0 adapter's canonical supportedInterfaces.

Attributes

capabilitiesAgentCapabilities

Always normalized to an AgentCapabilities instance after successful initialization.

interfaceslist[AgentInterface]

Always normalized to interface objects after successful initialization. Invalid members raise TypeError.

Independent defaults

Lists, dictionaries, capabilities, and interfaces use dataclass default factories. Instances do not share their mutable default containers even though the concise signature displays familiar empty values.

Examples

from protolink import AgentCard, AgentInterface, AgentSkill
from protolink.core.agent_card import AgentCapabilities

card = AgentCard(
    name="weather_agent",
    description="Provides current conditions and short-range forecasts.",
    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,
    ),
    skills=[
        AgentSkill(
            id="forecast",
            description="Forecast weather for a supplied location.",
        )
    ],
    interfaces=[
        AgentInterface(
            url="grpcs://api.example.com:9443",
            transport="grpc",
        )
    ],
)

AgentCard.to_dict​

methodprotolink.AgentCard.to_dict
source
to_dict() -> dict[str, Any]

Serialize the card into ProtoLink's native discovery-card dictionary. Nested capabilities and skills become dictionaries, field names that belong to the wire contract use camel case, and additional interfaces are omitted when the list is empty.

Returns

datadict[str, Any]

A new outer dictionary containing protocolVersion, inputFormats, outputFormats, securitySchemes, and optionally additionalInterfaces. Capability and skill dataclasses are recursively converted with dataclasses.asdict().

Native card behavior

The current serializer does not include role. It also returns the original tags, format lists, and security mapping rather than deep-copying those containers. The A2A 1.0 adapter uses a separate canonical serializer.

Examples

payload = card.to_dict()

print(payload["name"])                  # "weather_agent"
print(payload["protocolVersion"])       # installed ProtoLink version
print(payload["additionalInterfaces"])  # serialized alternate route

AgentCard.from_dict​

classmethodprotolink.AgentCard.from_dict
source
from_dict(
  data: dict[str, Any],
) -> AgentCard

Construct an AgentCard from the native discovery dictionary. This is the validated and normalizing path for data received from JSON, a registry, or another transport boundary.

Parameters

datadict[str, Any]required

Native card mapping. name, description, and url must be present and truthy. Capabilities and skills are read from nested mappings; wire-facing names such as protocolVersion and securitySchemes are converted to Python attribute names.

Returns

cardAgentCard

A new card with normalized AgentCapabilities, AgentSkill, and AgentInterface values.

Raises

ValueError

Raised when any mandatory identity field is absent or falsy, or when nested dataclass values cannot be constructed.

TypeError

Raised when nested values have incompatible shapes or interface members cannot be normalized.

KeyError

Raised by malformed interface mappings that do not contain their required url.

Accepted interface keys

Additional interfaces are read from interfaces first and then additionalInterfaces. The former is accepted for compatibility; to_dict() emits the latter.

Examples

data = {
    "name": "weather_agent",
    "description": "Weather service",
    "url": "https://api.example.com/weather",
    "capabilities": {"streaming": True},
    "additionalInterfaces": [
        {
            "url": "grpcs://api.example.com:9443",
            "transport": "grpc",
        }
    ],
}

card = AgentCard.from_dict(data)

AgentCard.get_prompt_format​

methodprotolink.AgentCard.get_prompt_format
source
get_prompt_format() -> str

Generate deterministic JSON metadata for an LLM delegation prompt. The object contains the agent's name, description, capabilities, and a skill-sorted tools array with each skill's description, schemas, and examples. It deliberately omits the transport URL so model output cannot select an arbitrary destination directly.

Returns

prompt_textstr

A valid, indented JSON object with stable key and skill ordering. Legacy Python type objects in schema metadata are rendered as strings rather than producing an invalid prompt document.

Prompt representation

This JSON is an LLM context representation, not the native discovery-card wire schema. Treat descriptions, schemas, examples, and capability values as untrusted metadata; use to_dict() for card serialization.

AgentCapabilities​

dataclassprotolink.core.agent_card.AgentCapabilities
source
class AgentCapabilities(
  streaming: bool = False,
  push_notifications: bool = False,
  state_transition_history: bool = False,
  delegation: bool = True,
  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,
  rag: bool = False,
  code_execution: bool = False,
)

Coarse capability and capacity declaration carried by an AgentCard. These values help peers choose an interaction mode; they do not themselves enable streaming, tools, delegation, or execution infrastructure.

Parameters

streamingbooldefault: False

Advertises that the agent can produce task events through a streaming-capable transport. The selected transport must also support streaming.

push_notificationsbooldefault: False

Advertises webhook or other push delivery for task updates. The flag is descriptive and does not configure a callback endpoint.

state_transition_historybooldefault: False

Indicates that detailed task lifecycle transitions can be provided to clients.

delegationbooldefault: True

Indicates that the agent may delegate work to other agents. It defaults to enabled in ProtoLink's native runtime profile.

has_llmbooldefault: False

Declares that an LLM is part of the agent's processing path. This does not expose the provider or model identifier.

max_concurrencyintdefault: 1

Advertised maximum simultaneous task capacity. No positivity validator runs in this dataclass; runtime schedulers decide how to enforce the declared value.

message_batchingbooldefault: False

Indicates support for processing multiple messages as one request.

tool_callingbooldefault: False

Indicates that the agent can invoke registered tools or external APIs.

multi_step_reasoningbooldefault: False

Advertises a multi-step reasoning or planning path.

timeout_supportbooldefault: False

Indicates that task or operation timeouts are understood by the agent.

ragbooldefault: False

Advertises retrieval-augmented generation support. Attaching a Knowledge source through the Agent constructor, add_knowledge(), or @agent.retriever sets this flag to true automatically and registers the corresponding read-only search tool. Setting the flag manually remains descriptive and does not create a retriever or index.

code_executionbooldefault: False

Advertises access to a code-execution facility. This flag is not a security boundary; the actual sandbox and policy must be configured separately.

Import path

AgentCapabilities is used by the public AgentCard, but it is not currently re-exported from protolink or protolink.models. Import it from protolink.core.agent_card.

AgentCapabilities.as_dict​

methodprotolink.core.agent_card.AgentCapabilities.as_dict
source
as_dict() -> dict[str, Any]

Convert every capability field into a plain dictionary using dataclasses.asdict().

Returns

capabilitiesdict[str, Any]

A new flat dictionary containing enabled and disabled booleans plus max_concurrency.

AgentCapabilities.enabled​

methodprotolink.core.agent_card.AgentCapabilities.enabled
source
enabled() -> list[str]

Produce a compact display list of truthy boolean capabilities and positive integer capacities.

Returns

nameslist[str]

Boolean fields appear by name. Positive integer fields appear as "field: value"; consequently the default profile includes delegation and max_concurrency: 1.

Examples

from protolink.core.agent_card import AgentCapabilities

capabilities = AgentCapabilities(streaming=True, max_concurrency=5)

print(capabilities.enabled())
# ["streaming", "delegation", "max_concurrency: 5"]

AgentSkill​

dataclassprotolink.AgentSkill
source
class AgentSkill(
  id: str,
  description: str = "",
  input_schema: dict[str, Any] = {},
  output_schema: dict[str, Any] = {},
  tags: list[str] = [],
  examples: list[Any] = [],
)

Task-level capability advertised by an agent. A skill combines a stable identifier with enough schema and example information for humans, registries, and delegating models to understand how to call it.

Parameters

idstrrequired

Human-readable operation identifier such as weather_forecast. The dataclass does not enforce uniqueness; cards and registries are responsible for avoiding ambiguous identifiers.

descriptionstrdefault: ""

Detailed explanation of what the skill does, when to use it, and any important boundaries. It is included in delegation prompt material.

input_schemadict[str, Any]default: {}

JSON Schema describing the accepted input payload. ProtoLink stores the mapping but does not validate that it is a complete or valid JSON Schema at construction time.

output_schemadict[str, Any]default: {}

JSON Schema describing the successful result payload.

tagslist[str]default: []

Search and categorization labels scoped to this skill.

exampleslist[Any]default: []

Representative inputs, outputs, or usage scenarios. Values may be strings or structured JSON-compatible objects.

None normalization

If tags, examples, input_schema, or output_schema is explicitly passed as None, post_init replaces it with a fresh empty list or dictionary. The identifier and description are not otherwise validated.

Examples

from protolink import AgentSkill

skill = AgentSkill(
    id="weather_forecast",
    description="Return a forecast for a named 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"},
    ],
)

AgentInterface​

frozen dataclassprotolink.AgentInterface
source
class AgentInterface(
  url: str,
  transport: TransportType,
  protocol_version: str = protolink_version,
)

Additional endpoint exposed by the same logical agent. The primary route remains AgentCard.url plus AgentCard.transport; use an interface only when one agent is genuinely reachable over another URL, transport, or protocol version.

Parameters

urlstrrequired

Absolute endpoint for the alternate interface. Syntax is preserved as supplied.

transportTransportTyperequired

Registered transport name for this endpoint. Unlike from_dict(), direct construction requires the argument explicitly.

protocol_versionstrdefault: protolink_version

Protocol version served specifically by this endpoint.

Immutable value

The dataclass is frozen and slot-backed. Assigning to an interface field after construction raises dataclasses.FrozenInstanceError.

AgentInterface.from_dict​

classmethodprotolink.AgentInterface.from_dict
source
from_dict(
  data: Mapping[str, Any],
) -> AgentInterface

Normalize an alternate-interface mapping from the native card wire format.

Parameters

dataMapping[str, Any]required

Mapping containing url and optionally transport and protocolVersion. The URL and protocol version are converted with str(); transport is preserved.

Returns

interfaceAgentInterface

A new immutable interface. Missing transport defaults to http; missing protocol version defaults to the installed ProtoLink version.

Raises

KeyError

Raised when the required url key is absent.

AgentInterface.to_dict​

methodprotolink.AgentInterface.to_dict
source
to_dict() -> dict[str, Any]

Serialize the interface using the field names expected inside AgentCard.additionalInterfaces.

Returns

datadict[str, Any]

Dictionary containing url, transport, and camel-cased protocolVersion.

Type aliases used by AgentCard​

MimeType​

MimeType enumerates the media types used by input_formats and output_formats. The annotation helps static checking and documentation; values are not runtime-validated by AgentCard.

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​

SecuritySchemeType enumerates the supported top-level security scheme categories. The nested configuration remains an application-supplied mapping.

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

Tasks and lifecycle​

A Task is the durable unit of work exchanged between agents. Messages record the conversation and control inputs, artifacts record produced outputs, and TaskState enforces how the work moves from submission to a terminal result.

Task​

dataclassprotolink.Task
source
class Task(
  id: str = generate_task_id(),
  state: TaskState = TaskState.SUBMITTED,
  messages: list[Message] = [],
  artifacts: list[Artifact] = [],
  metadata: dict[str, Any] = {},
  flow_state: dict[str, Any] = {},
  created_at: str = utc_now(),
)

State container for one agentic work unit. The model combines lifecycle state, chronological communication, produced artifacts, extensible metadata, and flow-local state. It also caches the most recently added message or artifact for constant-time access.

Parameters

idstrdefault: generate_task_id()

Unique task identifier. The default generator produces a task_-prefixed timestamp and random suffix. Supply an existing identifier when rehydrating or correlating work across systems.

stateTaskState | strdefault: TaskState.SUBMITTED

Current lifecycle state. During post_init, strings are converted with TaskState(value), so only exact enum values such as working or input-required succeed.

messageslist[Message]default: []

Ordered communication and control messages associated with the task. The constructor stores the supplied list and builds the last-item cache from its final element.

artifactslist[Artifact]default: []

Ordered outputs, previews, diagnostics, and resources produced during the run.

metadatadict[str, Any]default: {}

Extensible task metadata. Lifecycle helpers add state_history, error, or cancel_reason entries here.

flow_statedict[str, Any]default: {}

Flow and orchestration context carried with the task independently from general metadata.

created_atstrdefault: utc_now()

ISO 8601 UTC timestamp captured at construction time.

Attributes

is_terminalbool

Read-only property that is true for COMPLETED, CANCELED, and FAILED.

Last-item cache

Use add_message() and add_artifact() after construction. Mutating messages or artifacts directly does not refresh the private cache used by get_last_item() and get_last_part_content().

Raises

ValueError

Raised when a string state is not one of the exact TaskState values.

TypeError

Raised when state is neither a TaskState nor a string.

Task 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 cannot transition further. UNKNOWN is primarily a compatibility state; the current transition graph permits it to move to any enum value.

Every successful state change is recorded in task.metadata["state_history"] when that key is a list:

[
    {
        "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
Performance

add_message(), add_artifact(), update_state(), and cached last-item lookup are constant-time operations. Serialization remains proportional to the number of nested messages and artifacts.

Task.add_message​

methodprotolink.Task.add_message
source
add_message(
  message: Message,
) -> Task

Append a message to the task and make it the cached most recent item.

Parameters

messageMessagerequired

Communication or control message to append. The method does not perform an isinstance check, so callers should supply a real Message to preserve serialization and helper behavior.

Returns

selfTask

The same task instance, allowing fluent construction.

Side effect

Mutates messages and replaces the cached last item, but does not change task state or timestamps.

Examples

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

Task.add_artifact​

methodprotolink.Task.add_artifact
source
add_artifact(
  artifact: Artifact,
) -> Task

Append a durable output artifact and make it the cached most recent item.

Parameters

artifactArtifactrequired

Result, preview, diagnostic, or resource produced by the task.

Returns

selfTask

The mutated task for chaining.

Side effect

Mutates artifacts and the last-item cache. It does not automatically complete the task.

Examples

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

Task.update_state​

methodprotolink.Task.update_state
source
update_state(
  state: TaskState | str,
) -> Task

Move the task through the enforced lifecycle graph. Repeating the current state is a no-op; a successful change is recorded in metadata["state_history"].

Parameters

stateTaskState | strrequired

Destination state as an enum or exact serialized value. The method validates the transition from the task's current state before mutating it.

Returns

selfTask

The same task after a valid transition or repeated-state no-op.

Raises

ValueError

Raised for an unknown string value or a transition not present in the lifecycle graph. State and history remain unchanged when the graph check fails.

TypeError

Raised when the destination is neither an enum nor a string.

History behavior

If metadata["state_history"] already exists but is not a list, the state still changes and the transition record is silently skipped.

Examples

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

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

Task.begin​

methodprotolink.Task.begin
source
begin() -> Task

Mark the task as actively being processed. This is exactly update_state(TaskState.WORKING) and therefore follows the same transition rules and history behavior.

Returns

selfTask

The task in WORKING state.

Raises

ValueError

Raised when the current state cannot transition to WORKING, including terminal states.

Task.require_input​

methodprotolink.Task.require_input
source
require_input(
  message: Message | None = None,
) -> Task

Move the task to INPUT_REQUIRED and optionally append a message explaining what information is missing. A submitted task first moves through WORKING, which preserves a valid and observable lifecycle.

Parameters

messageMessage | Nonedefault: None

Optional prompt or status message appended after the state reaches INPUT_REQUIRED. Falsy values are ignored.

Returns

selfTask

The task in INPUT_REQUIRED state.

Raises

ValueError

Raised when the current lifecycle state cannot reach WORKING or INPUT_REQUIRED.

Repeated requests

Calling this method while already in INPUT_REQUIRED records two new transitions: back to WORKING, then to INPUT_REQUIRED.

Task.complete​

methodprotolink.Task.complete
source
complete(
  response_text: str,
) -> Task

Finish the task successfully and append the final text as an agent message. Submitted or input-required tasks first move through WORKING, so the convenience method can be used without manually creating the intermediate state.

Parameters

response_textstrrequired

Final response content. It is wrapped with Message.agent() after the state becomes COMPLETED.

Returns

selfTask

The completed task with the response message as its cached last item.

Raises

ValueError

Raised when the current state cannot transition through WORKING to COMPLETED.

Examples

task.complete("The weather is sunny and 24°C.")

print(task.state)  # TaskState.COMPLETED
print(task.get_last_part_content())  # "The weather is sunny and 24°C."

Task.fail​

methodprotolink.Task.fail
source
fail(
  error_message: str,
) -> Task

Move the task to FAILED and store a human-readable error in task metadata.

Parameters

error_messagestrrequired

Failure explanation stored at metadata["error"]. The method does not append an error part or response message.

Returns

selfTask

The failed task.

Raises

ValueError

Raised if the current state cannot transition to FAILED. The error metadata is written only after a successful transition.

Task.cancel​

methodprotolink.Task.cancel
source
cancel(
  reason: str | None = None,
) -> Task

Move the task model to CANCELED and optionally retain a reason. This updates lifecycle data only; it does not interrupt an operation that is currently executing.

Parameters

reasonstr | Nonedefault: None

Optional explanation stored at metadata["cancel_reason"]. Empty strings are treated as absent and are not stored.

Returns

selfTask

The canceled task.

Raises

ValueError

Raised when the current state cannot transition to CANCELED.

Running work

To interrupt active execution, use await agent.cancel_task(task.id, reason=...) or await client.cancel_task(agent_url, task.id, reason=...). See runtime cancellation.

Task.to_dict​

methodprotolink.Task.to_dict
source
to_dict() -> dict[str, Any]

Serialize a task and all nested messages and artifacts into the native transport shape.

Returns

datadict[str, Any]

New outer dictionary containing the string state value, serialized nested objects, metadata, flow state, identifier, and creation timestamp.

Copy semantics

Nested messages and artifacts are converted recursively. The task's metadata and flow_state mappings are attached directly rather than deep-copied.

Task.from_dict​

classmethodprotolink.Task.from_dict
source
from_dict(
  data: dict[str, Any],
) -> Task

Rehydrate a task from native serialized data, including nested Message, Part, and Artifact instances. Construction also rebuilds the cached last item by comparing the final message and artifact timestamps.

Parameters

datadict[str, Any]required

Task mapping. Missing fields receive constructor defaults; nested message and artifact lists are normalized by their respective from_dict() methods.

Returns

taskTask

A new task with enum state, hydrated nested content, and a reconstructed last-item cache.

Raises

ValueError

Raised when the serialized state is not a valid TaskState value or nested data fails value conversion.

KeyError or TypeError

Propagated from malformed nested part, message, or artifact payloads.

Examples

task = Task.from_dict(
    {
        "state": "working",
        "messages": [],
        "artifacts": [],
    }
)

print(task.state)  # TaskState.WORKING

Task.create​

classmethodprotolink.Task.create
source
create(
  message: Message,
) -> Task

Create a submitted task with one initial message and initialize the last-item cache without a second scan.

Parameters

messageMessagerequired

Initial user, agent, infer, tool-call, or other message.

Returns

taskTask

New SUBMITTED task whose messages contains exactly the supplied message.

Examples

task = Task.create(Message.user("Analyze this data"))

print(len(task.messages))  # 1
print(task.state)          # TaskState.SUBMITTED

Task.create_infer​

classmethodprotolink.Task.create_infer
source
create_infer(
  *,
  prompt: str | None = None,
  user: str | None = None,
  output_schema: dict[str, Any] | None = None,
  metadata: dict[str, Any] | None = None,
) -> Task

Create a submitted task containing one user-role message with an infer part. Agents interpret that part as a request to invoke their configured LLM.

Parameters

promptstr | Nonedefault: None

Main inference instruction. Omitted values are removed from the part payload rather than serialized as null.

userstr | Nonedefault: None

Optional user identity or user-specific context passed inside the infer payload. It does not change the enclosing message role.

output_schemadict[str, Any] | Nonedefault: None

Optional structured-output schema that the receiving agent may use when configuring inference.

metadatadict[str, Any] | Nonedefault: None

Additional infer-operation metadata stored inside the part, separate from Task.metadata.

Returns

taskTask

A new submitted task initialized through Message.infer().

Empty payload

All arguments are optional. Calling Task.create_infer() with no values still creates a valid infer part whose content is an empty dictionary.

Examples

task = Task.create_infer(
    prompt="Extract the invoice total.",
    output_schema={
        "type": "object",
        "properties": {"total": {"type": "number"}},
        "required": ["total"],
    },
)

Task.create_tool_call​

classmethodprotolink.Task.create_tool_call
source
create_tool_call(
  *,
  tool_name: str,
  args: dict[str, Any] | None = None,
  call_id: str | None = None,
) -> Task

Create a submitted task containing one user-role tool_call message. This is the direct task-level entry point for asking an agent to execute a registered tool without first asking its LLM to select one.

Parameters

tool_namestrrequired

Registered tool or capability name to invoke. Resolution happens when the receiving agent executes the task.

argsdict[str, Any] | Nonedefault: None

Keyword arguments for the tool. None and an empty dictionary both become a new empty argument mapping.

call_idstr | Nonedefault: None

Optional correlation identifier. If omitted, Part.tool_call() generates a tool_call_-prefixed identifier.

Returns

taskTask

New submitted task containing the generated tool-call message.

Examples

task = Task.create_tool_call(
    tool_name="get_weather",
    args={"location": "Athens"},
)

Task.get_last_item​

methodprotolink.Task.get_last_item
source
get_last_item() -> Message | Artifact | None

Return the message or artifact most recently cached by task construction, deserialization, add_message(), or add_artifact().

Returns

itemMessage | Artifact | None

Cached object, or None when the task has no messages or artifacts. For a task initialized with both lists, their final items are compared by timestamp.

Complexity

Lookup is O(1). The method does not rescan the lists, which is why direct list mutation can leave the result stale.

Task.tool_call​

staticmethodprotolink.Task.tool_call
source
tool_call(
  *,
  tool_name: str,
  args: dict[str, Any] | None = None,
  call_id: str | None = None,
) -> Part

Create a standalone tool_call part. This is a convenience alias for Part.tool_call(); it does not create or mutate a task.

Parameters

tool_namestrrequired

Tool or capability identifier.

argsdict[str, Any] | Nonedefault: None

Tool arguments; falsy values become an empty dictionary.

call_idstr | Nonedefault: None

Optional correlation identifier, otherwise generated automatically.

Returns

partPart

Typed tool-call part suitable for a message or task.

Task.infer​

staticmethodprotolink.Task.infer
source
infer(
  *,
  prompt: str | None = None,
  user: str | None = None,
  output_schema: dict[str, Any] | None = None,
  metadata: dict[str, Any] | None = None,
) -> Part

Create a standalone infer part without constructing a message or task. This delegates directly to Part.infer().

Parameters

promptstr | Nonedefault: None

Model instruction included only when non-None.

userstr | Nonedefault: None

Optional user context included only when non-None.

output_schemadict[str, Any] | Nonedefault: None

Optional structured-output schema.

metadatadict[str, Any] | Nonedefault: None

Optional infer-operation metadata.

Returns

partPart

Part with type infer and a dictionary containing only supplied values.

Task.get_last_part_content​

methodprotolink.Task.get_last_part_content
source
get_last_part_content() -> Any | None

Read the content of the final part on the cached most recent message or artifact. This is the concise result accessor used throughout examples and transport conformance tests.

Returns

contentAny | None

The final part's content, or None when there is no cached item or that item has no parts. Typed content such as ToolOutput is returned as the object, not automatically unwrapped to its result.

Task example

from protolink import Message, Task

task = Task.create(Message.user("What's the weather in New York?"))

task.begin()
task.complete("It's 22°C and sunny in New York.")

print(task.is_terminal)             # True
print(task.get_last_part_content()) # "It's 22°C and sunny in New York."

TaskState​

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

Enumeration of task lifecycle states. In-memory tasks hold enum members; Task.to_dict() serializes their string values.

Values

ValueMeaning
SUBMITTEDTask has been accepted but processing has not started.
WORKINGAgent is actively processing the task.
INPUT_REQUIREDAgent cannot continue without additional input.
COMPLETEDTask finished successfully.
CANCELEDTask was canceled before successful completion.
FAILEDTask ended because of an error.
UNKNOWNCompatibility state used when lifecycle status is not known.
Terminal states

COMPLETED, CANCELED, and FAILED intentionally have no outgoing transitions.

Server endpoints​

ProtoLink servers declare behavior through transport-neutral endpoint specifications. HTTP, WebSocket, gRPC, runtime-memory, and backend adapters consume the same declaration and decide how to bind a path, parse input, invoke the handler, and serialize its result.

EndpointSpec​

frozen dataclassprotolink.models.EndpointSpec
source
class EndpointSpec(
  name: str,
  path: str,
  method: HttpMethod,
  handler: Callable[..., Any],
  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",
)

Transport-agnostic declaration of one server endpoint. Server implementations assemble these values; transport backends use them to register routes and adapt raw requests without coupling agent or registry logic to FastAPI, Starlette, WebSocket, gRPC, or runtime-memory APIs.

Parameters

namestrrequired

Unique internal endpoint name used by transport routing tables and diagnostics. The dataclass does not enforce uniqueness; the server or backend that registers the collection is responsible for collisions.

pathstrrequired

Route path such as /tasks/. The transport backend interprets path syntax and route parameters.

methodHttpMethodrequired

HTTP-style method: GET, POST, DELETE, PUT, or PATCH. Non-HTTP transports use the value as part of the common routing contract.

handlerCallable[..., Any]required

Sync function, async function, or streaming callable invoked by the transport. Its expected argument depends on request_source and the optional parser; its return value is normalized by the selected backend.

content_type"json" | "html"default: "json"

Response rendering mode. HTTP backends return an HTML response only for html; otherwise they use their JSON normalization path.

streamingbooldefault: False

Compatibility flag indicating that the handler returns an async stream of events.

mode"request_response" | "stream"default: "request_response"

Explicit interaction mode. Current transports generally treat mode="stream" or streaming=True as a streaming declaration.

request_parserCallable[[Any], Any] | Nonedefault: None

Optional normalizer or validator applied to the selected raw request value before handler invocation. Parser exceptions propagate through the backend's normal error path.

request_sourceRequestSourceTypedefault: "none"

Selects the handler input: none, body, query_params, form, headers, path_params, or a transport-neutral request view.

Declaration, not validation

The frozen dataclass stores the values but does not verify path syntax, method literals, handler callability, or consistency between streaming and mode. Registration and request handling expose incompatible declarations.

Immutable specification

Endpoint specifications are frozen after construction. Create a replacement value when route behavior changes instead of mutating one registered with a transport.

Examples

from protolink.models import EndpointSpec

async def create_task(task):
    return await agent.execute_task(task)

endpoint = EndpointSpec(
    name="create_task",
    path="/tasks/",
    method="POST",
    handler=create_task,
    request_source="body",
)

For a stream, declare the interaction explicitly:

stream_endpoint = EndpointSpec(
    name="stream_task",
    path="/tasks/stream",
    method="POST",
    handler=agent.run_task_streaming,
    request_source="body",
    streaming=True,
    mode="stream",
)

LLM context models​

Agent protocol messages and LLM context have different jobs. Message and Part represent task-level communication; LLMMessage and ConversationHistory represent the compact, provider-neutral sequence translated into OpenAI, Anthropic, Gemini, local-server, or other model requests.

History compaction is a control-plane operation. HistoryCompactionRequest carries the requested strategy across direct Agent and client/server APIs, while HistoryCompactionResult reports exactly what changed.

LLMMessage​

slot dataclassprotolink.llms.history.LLMMessage
source
class LLMMessage(
  role: LLMMessageRole,
  content: str,
  name: str | None = None,
  metadata: dict[str, Any] = {},
  id: str = uuid4(),
  created_at: datetime = datetime.now(timezone.utc),
  tool_calls: dict[str, Any] = {},
  tool_name: str | None = None,
)

Canonical context entry used internally across all LLM providers. It keeps the provider-neutral role and text content alongside tracing metadata and optional provider-specific tool-call information.

Parameters

roleLLMMessageRolerequired

One of SYSTEM, USER, ASSISTANT, or TOOL. Direct construction expects an enum because to_dict() accesses role.value; use from_dict() to coerce serialized strings.

contentstrrequired

Provider-neutral textual content. Tool-call metadata may be stored separately, but content is still required by the dataclass.

namestr | Nonedefault: None

Optional function or tool name exposed by simplified provider message conversion.

metadatadict[str, Any]default: {}

Framework or application metadata retained by full-history serialization. It is not included in the simplified ConversationHistory.messages provider view.

idstrdefault: uuid4()

UUID string used for tracing and persistence.

created_atdatetimedefault: datetime.now(timezone.utc)

Timezone-aware UTC creation time. Full serialization converts it to ISO 8601.

tool_callsdict[str, Any]default: {}

Provider-specific tool-call payload retained for adapters and complete persistence.

tool_namestr | Nonedefault: None

Additional provider-specific tool name field. This is distinct from name and is preserved only by full serialization.

Import path

LLMMessage and LLMMessageRole are lower-level LLM context types, not top-level ProtoLink exports. Import them from protolink.llms.history.

LLMMessage.to_dict​

methodprotolink.llms.history.LLMMessage.to_dict
source
to_dict() -> dict[str, Any]

Serialize every canonical message field for persistence, compaction, copying, or telemetry.

Returns

datadict[str, Any]

Dictionary containing the string role, content, names, metadata, identifier, ISO timestamp, and tool-call payload.

Shallow containers

The outer dictionary is new, but metadata and tool_calls are returned by reference rather than deep-copied.

LLMMessage.from_dict​

classmethodprotolink.llms.history.LLMMessage.from_dict
source
from_dict(
  data: dict[str, Any],
) -> LLMMessage

Rehydrate a full serialized context message. This is the canonical path used by history copy, replacement, and persistence.

Parameters

datadict[str, Any]required

Mapping with required role and content. Optional metadata, tracing, and tool fields receive constructor defaults.

Returns

messageLLMMessage

New slot-backed message with an enum role and a parsed datetime when created_at is present.

Raises

KeyError

Raised when role or content is absent.

ValueError

Raised for an unknown role value or invalid ISO timestamp.

TypeError

Raised when a present timestamp has a value that datetime.fromisoformat() cannot consume.

ConversationHistory​

classprotolink.llms.history.ConversationHistory
source
class ConversationHistory(
  system_prompt: str | None = None,
)

Provider-agnostic conversation container backed by collections.deque. It provides fast appends and system-message prepends while keeping a complete serialization format for state persistence and a simplified format for model adapters.

Parameters

system_promptstr | Nonedefault: None

Optional first system instruction. A message is created only when the value is truthy, so None and an empty string both produce an initially empty history.

Attributes and protocols

messageslist[dict[str, Any]]

Read-only property returning a newly built simplified list with role, content, and optional name. It deliberately omits metadata, IDs, timestamps, tool calls, and tool_name.

len(history)int

Number of canonical messages currently stored.

iter(history)Iterable[LLMMessage]

Iterates the live deque in chronological order.

Import path

Import this lower-level model from protolink.llms.history. Most direct users encounter it through llm.history or llm.use_history().

Two serialization views

Use history.messages for simple provider input and history.to_list() for persistence, copying, or compaction. The latter preserves every LLMMessage field.

ConversationHistory.add_system​

methodprotolink.llms.history.ConversationHistory.add_system
source
add_system(
  content: str,
) -> None

Append a system-role message to the end of history.

Parameters

contentstrrequired

System instruction stored in a newly generated LLMMessage.

Append semantics

This method does not enforce that a system message is first or unique. Use set_system() to create or replace the leading system instruction, or reset_to_system() to discard all other history.

ConversationHistory.add_user​

methodprotolink.llms.history.ConversationHistory.add_user
source
add_user(
  content: str,
  **metadata: Any,
) -> None

Append a user-role context message.

Parameters

contentstrrequired

User text sent to provider adapters.

**metadataAny

Keyword metadata retained in full history. Passing metadata={"key": "value"} creates a nested key named metadata; pass key="value" when a flat metadata entry is intended.

ConversationHistory.add_assistant​

methodprotolink.llms.history.ConversationHistory.add_assistant
source
add_assistant(
  content: str,
  **metadata: Any,
) -> None

Append an assistant-role context message, optionally retaining framework metadata for persistence and telemetry.

Parameters

contentstrrequired

Assistant response text.

**metadataAny

Arbitrary keyword metadata stored on the canonical message.

ConversationHistory.add_tool​

methodprotolink.llms.history.ConversationHistory.add_tool
source
add_tool(
  content: str,
  tool_name: str,
  **metadata: Any,
) -> None

Append a tool-role response. The tool name is stored in LLMMessage.name so simplified provider conversion includes it.

Parameters

contentstrrequired

Tool response represented as text.

tool_namestrrequired

Name of the tool that produced the response.

**metadataAny

Additional canonical-message metadata.

ConversationHistory.add_raw​

methodprotolink.llms.history.ConversationHistory.add_raw
source
add_raw(
  message: dict[str, Any],
) -> None

Append a message from a simplified provider-style mapping.

Parameters

messagedict[str, Any]required

Mapping with required role, optional content, and optional tool_calls. Missing content becomes an empty string.

Raises

KeyError

Raised when role is absent.

ValueError

Raised when the role string is not a valid LLMMessageRole.

Lossy ingestion

This helper copies only role, content, and tool calls. Input keys such as name, metadata, ID, creation time, and tool name are ignored. Use replace() or from_list() with full message dictionaries when every canonical field must survive.

ConversationHistory.reset_to_system​

methodprotolink.llms.history.ConversationHistory.reset_to_system
source
reset_to_system(
  content: str,
) -> None

Discard every message and replace the history with one new system message.

Parameters

contentstrrequired

New system prompt. Unlike constructor initialization, an empty string is still stored as a system message.

Destructive mutation

The history object's identity remains stable, but all previous message objects become unreachable from it.

ConversationHistory.set_system​

methodprotolink.llms.history.ConversationHistory.set_system
source
set_system(
  content: str,
) -> None

Set the leading system instruction while preserving later conversation turns. If the first item is already a system message it is replaced; otherwise a new system message is prepended in constant time.

Parameters

contentstrrequired

New system prompt, including an empty string if that is explicitly desired.

Replacement identity

Replacing an existing system message creates a new LLMMessage, so its ID, creation time, metadata, and provider-specific fields are reset.

ConversationHistory.messages_raw​

methodprotolink.llms.history.ConversationHistory.messages_raw
source
messages_raw() -> list[LLMMessage]

Return a shallow list snapshot of the canonical message objects.

Returns

messageslist[LLMMessage]

New list in chronological order. The contained LLMMessage objects are shared with the history, so mutating one changes the canonical entry.

ConversationHistory.to_list​

methodprotolink.llms.history.ConversationHistory.to_list
source
to_list() -> list[dict[str, Any]]

Serialize the complete history for persistence, copying, or a lossless transformation.

Returns

messageslist[dict[str, Any]]

Chronological full-message dictionaries produced by LLMMessage.to_dict().

Full versus provider view

Unlike the messages property, this method preserves metadata, tracing IDs, timestamps, tool calls, and tool names.

ConversationHistory.copy​

methodprotolink.llms.history.ConversationHistory.copy
source
copy() -> ConversationHistory

Create an independent history by round-tripping every canonical message through full serialization.

Returns

historyConversationHistory

New history object with newly constructed LLMMessage instances that preserve all serialized fields.

Container depth

Message objects and their top-level dictionaries are recreated. Arbitrary nested objects inside metadata or tool-call mappings may still be shared because the message serializer is not a general deep-copy routine.

ConversationHistory.replace​

methodprotolink.llms.history.ConversationHistory.replace
source
replace(
  messages_data: Iterable[dict[str, Any]],
) -> None

Replace every canonical message while preserving the ConversationHistory object's identity. History compaction uses this behavior so LLMs, agents, and state modules can keep existing references to the same history container.

Parameters

messages_dataIterable[dict[str, Any]]required

Full chronological message dictionaries, normally from to_list(). The iterable is consumed once and each item is rehydrated through LLMMessage.from_dict().

Raises

KeyError or ValueError

Propagated from the first malformed serialized message. The new deque is built before assignment, so the existing history remains intact if hydration fails.

ConversationHistory.from_list​

classmethodprotolink.llms.history.ConversationHistory.from_list
source
from_list(
  messages_data: list[dict[str, Any]],
) -> ConversationHistory

Restore a new conversation from full serialized messages.

Parameters

messages_datalist[dict[str, Any]]required

Chronological full-message dictionaries.

Returns

historyConversationHistory

New history with one canonical LLMMessage per dictionary.

Raises

KeyError or ValueError

Propagated from malformed role, content, or timestamp fields.

ConversationHistory.truncate​

methodprotolink.llms.history.ConversationHistory.truncate
source
truncate(
  max_messages: int,
) -> None

Trim older history while preserving the first stored message and the newest suffix. The low-level operation mutates the live deque in place and is intended for histories whose first message is the system prompt.

Parameters

max_messagesintrequired

Maximum retained message count, including the protected first item. Values below two are rejected. When the history is already within the limit, no mutation occurs.

Raises

ValueError

Raised when max_messages is less than two.

Current first-message behavior

The implementation protects the first stored message without checking its role. If a history has no leading system message, its first user or assistant message is retained as though it were the system prompt.

Prefer structured compaction

For explicit recent-message limits, token budgets, summaries, and before/after reports, use LLM.compact_history().

History examples

from protolink.llms.history import ConversationHistory

history = ConversationHistory("You are a concise support assistant.")
history.add_user("My account is locked.", customer_id="customer-42")
history.add_assistant("I can help you recover access.")
history.add_tool(
    '{"recovery_email_sent": true}',
    tool_name="send_recovery_email",
)

persisted = history.to_list()
restored = ConversationHistory.from_list(persisted)

# Update the prompt without discarding the conversation.
restored.set_system("You are a concise, security-aware support assistant.")

HistoryCompactionRequest​

frozen dataclassprotolink.HistoryCompactionRequest
source
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 requests context maintenance without creating a task part, adding text to model history, or exposing compaction as an LLM tool.

Parameters

strategy"recent" | "tokens" | "summary"default: "recent"

Compaction algorithm. recent keeps a bounded newest suffix, tokens keeps a newest suffix under a soft estimated-token ceiling, and summary replaces older turns with one generated summary.

max_messagesintdefault: 20

Retained-message limit for the recent strategy, including a leading system message when one exists.

max_tokensintdefault: 4000

Estimated-token ceiling for the tokens strategy. Protected system and recent messages can make the result exceed this soft ceiling.

preserve_recentintdefault: 6

Number of newest non-system messages protected verbatim by tokens and summary.

summary_max_tokensintdefault: 512

Requested approximate maximum length of a generated summary.

session_idstr | Nonedefault: None

Optional persistent conversation session to load, compact, and save when the agent uses conversation state.

metadatadict[str, Any] | Nonedefault: None

Application metadata for logs or future control policy. The compactor itself ignores it.

Validation timing

Direct dataclass construction does not validate strategy names or numeric limits. from_dict() validates the strategy and normalizes numeric fields; the compactor validates strategy-specific limits when the operation runs.

Control plane

This immutable value is not shown to the model and does not consume prompt tokens.

HistoryCompactionRequest.to_dict​

methodprotolink.HistoryCompactionRequest.to_dict
source
to_dict() -> dict[str, Any]

Serialize the complete request for a transport body.

Returns

datadict[str, Any]

Dictionary containing all request fields. When the instance's metadata is None, the serialized value is normalized to an empty dictionary.

Frozen does not mean deeply immutable

The request fields cannot be reassigned, but a metadata dictionary supplied by the caller remains mutable.

HistoryCompactionRequest.from_dict​

classmethodprotolink.HistoryCompactionRequest.from_dict
source
from_dict(
  data: dict[str, Any],
) -> HistoryCompactionRequest

Normalize a JSON-compatible compaction request received through the control plane.

Parameters

datadict[str, Any]required

Request mapping. Missing values receive dataclass defaults. Numeric fields are converted with int(), non-None session IDs with str(), and metadata with dict().

Returns

requestHistoryCompactionRequest

New immutable request. Missing or falsy metadata is normalized to a fresh empty dictionary rather than None.

Raises

ValueError

Raised when strategy is not recent, tokens, or summary, or when a numeric value cannot be converted to an integer.

TypeError

Raised when numeric or metadata values have incompatible runtime shapes.

Range checks

Integer conversion is not range validation. Negative or otherwise invalid limits are rejected later by the compaction operation.

Examples

from protolink import HistoryCompactionRequest

request = HistoryCompactionRequest(
    strategy="tokens",
    max_tokens=8_000,
    preserve_recent=6,
    session_id="customer-42",
    metadata={"reason": "context pressure"},
)

HistoryCompactionResult​

frozen dataclassprotolink.HistoryCompactionResult
source
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 after direct LLM compaction, Agent control-plane compaction, or the equivalent client request. It records both message-count and estimated-token effects without requiring callers to diff histories manually.

Parameters

strategy"recent" | "tokens" | "summary"required

Strategy used for this attempt.

before_messagesintrequired

Canonical message count before compaction.

after_messagesintrequired

Canonical message count after compaction.

removed_messagesintrequired

Number of source messages removed. Summary replacement reports the number of older source messages represented by the summary.

before_tokensintrequired

Provider-neutral estimated token count before compaction.

after_tokensintrequired

Estimated token count after compaction.

summary_createdbooldefault: False

Whether summary compaction successfully inserted a generated summary.

Attributes

changedbool

Computed property that is true when at least one source message was removed or a summary was created.

Observational value

The result is immutable and does not retain the history itself. Counts are supplied by the compactor; direct constructor calls do not validate their consistency or non-negativity.

HistoryCompactionResult.to_dict​

methodprotolink.HistoryCompactionResult.to_dict
source
to_dict() -> dict[str, Any]

Serialize the compaction report for task results, telemetry, logging, or a client response.

Returns

datadict[str, Any]

Dictionary containing every dataclass field plus the computed changed property.

HistoryCompactionResult.from_dict​

classmethodprotolink.HistoryCompactionResult.from_dict
source
from_dict(
  data: dict[str, Any],
) -> HistoryCompactionResult

Create a compaction report from a JSON-compatible response.

Parameters

datadict[str, Any]required

Result mapping. Missing strategy defaults to recent; missing counts default to zero; missing summary status defaults to false. An incoming changed key is ignored because the property is recomputed.

Returns

resultHistoryCompactionResult

New immutable report with integer-normalized counts and a Boolean-normalized summary flag.

Raises

ValueError or TypeError

Raised when a count cannot be converted with int(). Strategy values and relationships between counts are not validated here.

Examples

report = llm.compact_history(
    "summary",
    preserve_recent=8,
    summary_max_tokens=600,
)

if report.changed:
    print(
        f"Compacted {report.before_messages} messages "
        f"to {report.after_messages}"
    )

print(report.to_dict())

See also​

  • LLMs - inference, history ownership, and compaction behavior.
  • Agents - task execution and lifecycle integration.
  • Flows - structured route decisions and task propagation.
  • State - conversation and task persistence.
  • Transport - endpoint binding and model serialization across backends.