Client

The Client layer in Protolink provides a high-level interface for agent-to-agent communication. It abstracts transport details and offers convenient methods for sending tasks, messages, and retrieving agent metadata.

AgentClient

The AgentClient is the primary entry point for programmatic agent interactions. It wraps a transport and provides a unified interface for communicating with Protolink agents.

The distinction is useful because application code should think in Agent operations such as “send this task” or “cancel that task,” not in HTTP headers, WebSocket frames, or gRPC metadata. AgentClient chooses the operation contract and parses the result; the selected transport only maps that contract onto its wire protocol. Changing from HTTP to gRPC therefore does not require rewriting task-level client code.

AgentClient follows the same progressive-control rule as Agent and Registry: a transport name creates a default client quickly, while a concrete transport carries TLS, limits, retries, keepalive, and protocol-specific behavior. The read-only client.transport property exposes the resolved transport for health and metric inspection.

By default, AgentClient uses ProtoLink's native request contract. AgentClient(..., a2a=True) requires HTTP and adds outbound A2A 1.0 discovery and translation. It does not remove the native methods: protocol="auto" prefers a native ProtoLink peer and selects A2A only for an A2A-only peer, while "protolink" and "a2a" are explicit choices. Advertised A2A interfaces must share the discovered card's origin unless the application explicitly sets a2a_allow_cross_origin=True for a trusted split-origin deployment.

from protolink import RetryPolicy, TransportConfig
from protolink.client import AgentClient
from protolink.transport import GRPCTransport

transport = GRPCTransport(
    url="grpc://127.0.0.1:0",
    config=TransportConfig(retry=RetryPolicy(max_attempts=3)),
)
client = AgentClient(transport)
print(client.transport.metrics)

Client moduleAgentClient

The typed application-facing client for sending tasks, messages, streaming requests, control-plane operations, registry calls, and LLM history actions over any supported transport.

protolink.client.AgentClient

Transport-backedAsync firstSync wrapper availableStreaming-awareControl-plane specs

Submit workSend task and message payloads to a remote agent while keeping transport details behind the client.send_task()

Stream progressYield live task events for SSE, WebSocket, JSON-RPC, and runtime transports.send_task_streaming()

Control runtimeCancel active work, inspect state, mutate state, and compact history through explicit request specs.cancel_task()

Use scriptsAccess a blocking facade for CLIs, notebooks, and simple orchestration scripts.client.sync

Design Philosophy: Async vs Sync

Protolink's client architecture exposes two APIs to accommodate different workflows:

1. Async API (Recommended): The core implementation. Ideal for modern applications, web servers (e.g., FastAPI), and high-performance multi-agent orchestration where non-blocking I/O is crucial.
2. Sync API (client.sync): A thin, blocking wrapper over the async methods. Designed for simple scripts, CLI tools, and environments where managing an asyncio event loop is cumbersome.

Async Loop Constraint

The Sync API (client.sync) uses asyncio.run() under the hood. It cannot be used inside an already running event loop (e.g., inside an async function). If you are inside an async def, always use the standard Async API.

Quick Start

from protolink.client import AgentClient
from protolink.models import Task

# Create a client (transport type + URL)
client = AgentClient(transport="http", url="http://localhost:8000")

# Create a task with an inference request
task = Task.create_infer(prompt="Book me a vacation to Santorini")

# Send to a remote agent
result = await client.send_task(agent_url="http://localhost:8010", task=task)

# Get the response
print(result.get_last_part_content())

Constructor

classprotolink.client.AgentClient

source↗

AgentClient(
  transport: Transport | TransportType,
  url: str | None = None,
  timeout: int = 300,
  *,
  a2a: bool = False,
  a2a_allow_cross_origin: bool = False,
) -> None

Create a high-level Agent client around one concrete transport. Construction also creates the per-instance blocking facade and, when requested, an A2A JSON-RPC adapter and bounded protocol-selection cache.

Parameters

transportTransport | TransportTyperequired

Configured transport instance or registered alias such as "http", "websocket", "sse", "json-rpc", "sse-json-rpc", "grpc", or "runtime". Existing instances are used directly and retain ownership of TLS, retries, limits, keepalive, and metrics.

urlstr | Nonedefault: None

Base address supplied to the transport factory when transport is an alias. It is ignored for an existing transport object.

timeoutintdefault: 300

Factory timeout in seconds for an alias-created transport. It does not overwrite a configured transport instance.

a2abooldefault: False

Enable outbound A2A card discovery, interface validation, task translation, and cancellation mapping while retaining native ProtoLink calls.

a2a_allow_cross_originbooldefault: False

Trust a standard Agent Card whose selected JSON-RPC interface has another origin. Keep the default unless that split-origin deployment is explicitly trusted.

Attributes

transportTransport

Read-only resolved transport.

a2abool

Read-only A2A-enabled flag.

syncSyncAgentClient

Blocking facade bound to this client.

A2A transport requirement

When a2a=True, construction immediately validates that the resolved transport's transport_type is exactly "http" and raises ValueError otherwise.

Use a string alias when ProtoLink should create a client transport with defaults. Use an existing transport object when the application needs TLS, production limits, retries, protocol-specific constructor options, or ownership of that exact instance. AgentClient never copies settings onto the transport and never creates a second hidden transport.

Examples:

# Simple: construct by transport name
client = AgentClient(transport="http", url="http://localhost:8000", timeout=120)

# Add A2A 1.0 outbound interoperability while retaining native calls
a2a_client = AgentClient(
    transport="http",
    url="http://localhost:8000",
    a2a=True,
)

# Only for a trusted deployment whose card intentionally advertises another origin
split_origin_client = AgentClient(
    transport="http",
    url="https://discovery.example",
    a2a=True,
    a2a_allow_cross_origin=True,
)

# Advanced: configure the transport first
from protolink import TLSConfig, TransportConfig
from protolink.transport import HTTPTransport

transport = HTTPTransport(
    url="https://agent.internal:8443",
    tls=TLSConfig(cafile="certs/ca.pem"),
    config=TransportConfig(shutdown_timeout=10),
)
client = AgentClient(transport=transport)

Transport Inspection

The read-only transport property exposes the concrete transport used by the client. This is the supported path for health, readiness, capability, and metric inspection:

snapshot = client.transport.metrics
print(snapshot.requests_succeeded, snapshot.retries)

health = client.transport.health()
print(health["status"], health["ready"])

See the Shared Transport API Reference for every configuration field, snapshot counter, exception type, and lifecycle probe.

---

Core Methods

send_task()

Sends a Task to a remote agent and returns the processed result.

async methodprotolink.client.AgentClient.send_task

source↗

async send_task(
  agent_url: str,
  task: Task,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Submit a complete task through the native request spec or the optional A2A adapter and wait for the processed Task result.

Parameters

agent_urlstrrequired

Peer base URL or transport-specific URI.

taskTaskrequired

Caller-created Task. Native submission is explicitly idempotent; A2A SendMessage is non-idempotent because the peer assigns its task ID.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

"protolink" skips discovery, "a2a" requires enabled A2A and validates a standard card, and "auto" probes the native card before falling back to A2A only on 404 or 405.

Returns

taskTask

Remote task state, messages, artifacts, and metadata normalized into ProtoLink's model.

Raises

ValueError

Invalid protocol value.

RuntimeError

A2A was explicitly requested but disabled.

transport or A2A error

Discovery, origin validation, authentication, timeout, network, translation, and remote failures propagate.

Example:

from protolink.models import Task

# Create an infer task
task = Task.create_infer(prompt="What's the weather in Athens?")

# Send and get result
result = await client.send_task("http://localhost:8010", task)
print(result.get_last_part_content())

For the common case where the task contains a single inference request, use send_infer_task() to create and submit the task in one call:

result = await client.send_infer_task(
    "What's the weather in Athens?",
    "http://localhost:8010",
)

With A2A enabled, automatic selection performs discovery before task submission. It probes /.well-known/agent.json first and falls back to /.well-known/agent-card.json only after a 404 or 405; authentication, network, timeout, and server failures are propagated rather than retried through another protocol. The process-local selection cache holds at most 1,024 peers, expires entries after five minutes, and removes the oldest entries when full. A2A SendMessage itself is non-idempotent and is not automatically retried because the remote server assigns the task ID.

client = AgentClient(transport="http", url="http://localhost:8000", a2a=True)

automatic = await client.send_task(peer_url, task)
a2a_only = await client.send_task(peer_url, task, protocol="a2a")
native_only = await client.send_task(peer_url, task, protocol="protolink")

Outbound A2A translation keeps the caller's local task ID and records the remote task ID, context, state, status timestamp, and agent URL in namespaced metadata. A fresh task does not send its local ID as an A2A taskId; that field is used only when continuing work previously returned by the same remote peer. The process-local local-to-remote mapping holds at most 1,024 tasks for one hour and removes expired or oldest entries. Continuation and A2A cancellation require a live mapping in the same client process.

---

send_infer_task()

Creates a task containing one inference request, sends it to a remote agent, and returns the processed task.

async methodprotolink.client.AgentClient.send_infer_task

source↗

async send_infer_task(
  query: str,
  agent_url: str,
  *,
  user: str | None = None,
  output_schema: dict[str, Any] | None = None,
  metadata: dict[str, Any] | None = None,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Build a new Task through Task.create_infer(prompt=query, ...), delegate it to send_task(), and return the complete remote task result.

Parameters

querystrrequired

Prompt placed in the task's inference part.

agent_urlstrrequired

Peer base URL or transport-specific URI.

userstr | Nonedefault: None

Optional user identifier or context included in the inference part.

output_schemadict[str, Any] | Nonedefault: None

Optional schema describing the expected model output.

metadatadict[str, Any] | Nonedefault: None

Optional metadata attached to the inference part.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Same native/A2A selection used by send_task().

Returns

taskTask

Processed task including remote state, messages, artifacts, and metadata.

result = await client.send_infer_task(
    query="Extract the invoice total",
    agent_url="http://localhost:8010",
    output_schema={
        "type": "object",
        "properties": {"total": {"type": "number"}},
        "required": ["total"],
    },
    metadata={"document_id": "invoice-42"},
)

Use send_task() directly when the task needs multiple messages, multiple parts, continuation state, or caller-defined task metadata.

---

send_task_streaming()

Sends a task and yields streamed events as they arrive. This is the public client API for live task progress, LLM chunks, tool events, and final task completion.

async generatorprotolink.client.AgentClient.send_task_streaming

source↗

send_task_streaming(
  agent_url: str,
  task: Task,
) -> AsyncIterator[Any]

Delegate a live task subscription to the configured transport. Unlike send_task(), this method has no protocol selector and uses ProtoLink's native streaming contract.

Parameters

agent_urlstrrequired

Peer address understood by the transport.

taskTaskrequired

Task serialized into the subscription request.

Yields

eventAny

Transport-decoded status, progress, LLM, artifact, or error event. Backends may yield dictionaries or typed objects.

Raises

NotImplementedError

The transport does not advertise streaming or implement subscribe().

transport or remote error

Subscription failures propagate while iterating.

Transport Support

Requires a transport that advertises streaming support and implements subscribe(). Supported choices include "sse", "json-rpc", "grpc", "websocket", and "runtime". Plain "http" remains request/response only and raises NotImplementedError.

Example with SSE JSON-RPC:

from protolink.client import AgentClient
from protolink.models import Task

client = AgentClient(transport="sse", url="http://localhost:8000")
task = Task.create_infer(prompt="Write a short haiku about agents")

async for event in client.send_task_streaming("http://localhost:8010", task):
    if event.get("type") == "task_llm_stream":
        print(event.get("content") or "", end="", flush=True)
    if event.get("final"):
        print("\nstream complete")

Applications that need a stable UI or replay contract can normalize these transport events with RunEvent.from_task_event(...) or record them through InMemoryEventSink. See Runtime for the versioned run-event envelope.

SSE, WebSocket, and gRPC transports recursively convert nested Protolink models and dataclasses into JSON-compatible values. Tool and delegated-agent events therefore preserve structured results such as ToolOutput inside content or metadata; clients do not need a custom encoder for these framework event payloads.

---

cancel_task()

Requests best-effort cancellation of a task currently executing on an agent and returns the task after the request is accepted.

async methodprotolink.client.AgentClient.cancel_task

source↗

async cancel_task(
  agent_url: str,
  task_id: str,
  *,
  reason: str | None = None,
  metadata: dict[str, Any] | None = None,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Send a control-plane cancellation request for active remote work. Native cancellation carries the caller-created task ID; A2A cancellation requires this client process to know the server-assigned ID mapping.

Parameters

agent_urlstrrequired

Peer running the task.

task_idstrrequired

Local ProtoLink task ID.

reasonstr | Nonedefault: None

Human-readable reason propagated into task cancellation metadata.

metadatadict[str, Any] | Nonedefault: None

Additional cancellation metadata; A2A translation carries it on the TaskId request.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Native, mapped A2A, or automatic selection.

Returns

taskTask

Remote task after cancellation acceptance.

Raises

ValueError

Invalid protocol.

RuntimeError | A2AClientError

A2A is disabled or no safe remote mapping exists.

remote error

Unknown or terminal tasks and transport failures propagate.

Best effort

Acceptance cannot roll back completed external side effects or forcibly interrupt synchronous code.

The task ID is known before submission because the caller creates the Task. Cancellation should be sent from another coroutine or control handler after the task has been accepted, usually after the first streamed status or progress event.

import asyncio

task = Task.create_infer(prompt="Perform long-running work")
running = asyncio.create_task(client.send_task(agent_url, task))

# Wait for application-specific acceptance or progress before canceling.
await task_started.wait()
canceled = await client.cancel_task(
    agent_url,
    task.id,
    reason="Stopped by the user",
    metadata={"source": "cli"},
)
result = await running

assert canceled.state.value == "canceled"
assert result.state.value == "canceled"

For native tasks, cancel_task() uses POST /tasks/cancel over HTTP, SSE JSON-RPC, WebSocket, gRPC, and RuntimeTransport. For a task previously returned through this client's A2A adapter, protocol="auto" uses the stored local-to-remote ID mapping and sends canonical A2A CancelTask; protocol="a2a" selects that path explicitly. The optional reason and metadata are translated into A2A cancellation metadata and reconstructed by a ProtoLink A2A server.

A blocking outbound A2A SendMessage does not reveal its server-assigned task ID until a response is returned, so it cannot be canceled through this client while that initial call is still blocked. An A2A task unknown to this client has no safe local-to-remote mapping and raises A2AClientError. In "auto", the client confirms an A2A peer and raises instead of sending the local ID to the native cancellation route. Cancellation remains a control-plane request: WebSocket sends native cancellation over a separate connection so it does not queue behind the active task stream.

Cancellation is intentionally best-effort. Async work normally stops at an await boundary; synchronous work and external systems may need their own cooperative cancellation or rollback mechanism. See Runtime cancellation for lifecycle, custom-handler, and side-effect guidance.

---

compact_history()

Requests LLM conversation-history compaction from an agent over the control plane.

async methodprotolink.client.AgentClient.compact_history

source↗

async compact_history(
  agent_url: str,
  *,
  strategy: HistoryCompactionStrategy = "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,
) -> HistoryCompactionResult

Request explicit LLM-history reduction through the non-idempotent control channel.

Parameters

agent_urlstrrequired

Target Agent address.

strategyLiteral["recent", "tokens", "summary"]default: "recent"

History reduction algorithm.

max_messagesintdefault: 20

Message-count budget for recent compaction.

max_tokensintdefault: 4000

Estimated token budget.

preserve_recentintdefault: 6

Recent messages protected from summarization.

summary_max_tokensintdefault: 512

Requested summary output budget.

session_idstr | Nonedefault: None

Persisted conversation session to load, compact, and save; omission targets the Agent LLM's current history.

metadatadict[str, Any] | Nonedefault: None

Application context attached to the control request.

Returns

resultHistoryCompactionResult

Before/after counts and strategy metadata.

This uses the built-in COMPACT_HISTORY_REQUEST spec (POST /llm/history/compact). It does not send a Task, does not create a model-visible tool, and does not add anything to the LLM prompt.

report = await client.compact_history(
    agent_url,
    strategy="tokens",
    max_tokens=8_000,
    preserve_recent=6,
    session_id="customer-42",
)

When the target agent has state=["conversation"] and session_id is supplied, the agent loads that session history, compacts it, and saves it back.

---

State Control Plane

Inspect, reset, or compact a remote agent's persistent state without sending a model-visible task.

AgentClient.describe_state

async methodprotolink.client.AgentClient.describe_state

source↗

async describe_state(
  agent_url: str,
  *,
  session_id: str | None = None,
  stores: tuple[str, ...] | list[str] | None = None,
  include_data: bool = False,
  metadata: dict[str, Any] | None = None,
) -> StateOperationResult

Inspect selected persistent stores through an idempotent control request.

Parameters

agent_urlstrrequired

Target Agent.

session_idstr | Nonedefault: None

Optional logical session.

storestuple[str, ...] | list[str] | Nonedefault: None

Requested store names; omission serializes an empty tuple for the server's default scope.

include_databooldefault: False

Include store data when supported, not only existence and counts.

metadatadict[str, Any] | Nonedefault: None

Request context.

Returns

resultStateOperationResult

Per-store enabled, missing, count, data, and error reports.

AgentClient.reset_state

async methodprotolink.client.AgentClient.reset_state

source↗

async reset_state(
  agent_url: str,
  *,
  session_id: str | None = None,
  stores: tuple[str, ...] | list[str] | None = None,
  metadata: dict[str, Any] | None = None,
) -> StateOperationResult

Request a non-idempotent persistent-state reset. A session ID narrows conversation deletion; without one, the server may reset its enabled namespace.

Parameters

agent_urlstrrequired

Target Agent.

session_idstr | Nonedefault: None

Optional conversation session to clear.

storestuple[str, ...] | list[str] | Nonedefault: None

Requested stores; omission serializes an empty tuple for server-side default scope.

metadatadict[str, Any] | Nonedefault: None

Control-request metadata.

Returns

resultStateOperationResult

Structured per-store reset and error report.

Destructive operation

The remote Agent's policy and approval handler authorize the actual mutation. The client does not ask for confirmation itself.

AgentClient.compact_state

async methodprotolink.client.AgentClient.compact_state

source↗

async compact_state(
  agent_url: str,
  *,
  session_id: str,
  strategy: HistoryCompactionStrategy = "tokens",
  max_messages: int = 20,
  max_tokens: int = 4000,
  preserve_recent: int = 6,
  summary_max_tokens: int = 512,
  metadata: dict[str, Any] | None = None,
) -> StateOperationResult

Compact one required persisted conversation session and return its structured state report. The request always selects the "conversation" store.

Parameters

agent_urlstrrequired

Target Agent.

session_idstrrequired

Existing persisted session.

strategyLiteral["recent", "tokens", "summary"]default: "tokens"

Reduction strategy.

max_messagesintdefault: 20

Message limit.

max_tokensintdefault: 4000

Estimated token limit.

preserve_recentintdefault: 6

Protected recent messages.

summary_max_tokensintdefault: 512

Summary budget.

metadatadict[str, Any] | Nonedefault: None

Control-request metadata.

Returns

resultStateOperationResult

Compaction, missing-session, disabled-store, and error reports.

state = await client.describe_state(
    agent_url,
    session_id="customer-42",
)

reset = await client.reset_state(
    agent_url,
    session_id="customer-42",
)

compacted = await client.compact_state(
    agent_url,
    session_id="customer-42",
    strategy="tokens",
    max_tokens=8_000,
)

These methods return StateOperationResult. They use control-channel request specs: DESCRIBE_STATE_REQUEST (POST /state/describe), RESET_STATE_REQUEST (POST /state/reset), and COMPACT_STATE_REQUEST (POST /state/compact).

---

send_message()

Convenience wrapper that creates a Task from a Message, sends it, and returns the response message.

async methodprotolink.client.AgentClient.send_message

source↗

async send_message(
  agent_url: str,
  message: Message,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Message

Wrap one Message in a new Task, delegate to send_task(), then return the newest agent or assistant message.

Parameters

agent_urlstrrequired

Peer address.

messageMessagerequired

Input role and parts.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Same native/A2A selection as send_task().

Returns

messageMessage

Last response whose role is "agent" or "assistant".

Raises

RuntimeError

The returned task has artifacts but no response message, or no response message at all.

Lossy convenience

Use send_task() when callers need task state, artifacts, context, metadata, or non-message results.

Example:

from protolink.models import Message

response = await client.send_message(
    agent_url="http://localhost:8010",
    message=Message.user("Hello, agent!")
)
print(response.parts[0].content)

send_message() requires a response message. If an A2A peer returns a completed task containing artifacts but no agent message, this convenience method raises RuntimeError; call send_task() to receive the full task and inspect its artifacts.

---

get_agent_card()

Retrieves the public AgentCard from a remote agent. Useful for discovery and capability inspection.

async methodprotolink.client.AgentClient.get_agent_card

source↗

async get_agent_card(
  agent_url: str,
) -> AgentCard

Fetch ProtoLink's native well-known card through an idempotent GET request and parse it into AgentCard.

Parameters

agent_urlstrrequired

Peer base address.

Returns

cardAgentCard

Validated identity, skills, interfaces, security, and capabilities.

Example:

card = await client.get_agent_card("http://localhost:8010")
print(f"Agent: {card.name}")
print(f"Description: {card.description}")
print(f"Skills: {[s.id for s in card.skills]}")

get_agent_card() reads ProtoLink's native card. When A2A is enabled, use get_a2a_agent_card() to fetch and validate the standard card and its JSON-RPC 1.0 interface:

AgentClient.get_a2a_agent_card

async methodprotolink.client.AgentClient.get_a2a_agent_card

source↗

async get_a2a_agent_card(
  agent_url: str,
) -> dict[str, Any]

Discover and validate the standard A2A 1.0 Agent Card, including its selected JSON-RPC interface and origin policy, then return a plain dictionary copy.

Parameters

agent_urlstrrequired

Base URL of the peer whose standard card should be discovered. Discovery starts from the well-known A2A card location and validates the advertised interface before returning.

Returns

carddict[str, Any]

Validated standard A2A 1.0 Agent Card copied into a plain dictionary, including its supported interfaces and capabilities.

Raises

RuntimeError

A2A was not enabled at construction.

A2AClientError

Discovery, schema, compatible-interface, or origin validation fails.

card = await client.get_a2a_agent_card("http://localhost:8010")
print(card["supportedInterfaces"])

---

Synchronous API

The AgentClient provides synchronous versions of its core methods for use in non-async contexts (scripts, notebooks, CLI tools). These are accessible via the client.sync property.

Internally, these methods use asyncio.run() to handle the asynchronous transport logic.

Do Not Use in Async Loops

The synchronous API should NOT be used inside an active event loop (e.g., inside FastAPI endpoints or async Jupyter cells) as it uses asyncio.run(), which will raise a RuntimeError.

The facade mirrors the asynchronous client method by method. Its arguments, return values, protocol selection, and control-plane behavior are identical; only the calling convention changes from await to a blocking call.

SyncAgentClient.send_task

methodprotolink.client.SyncAgentClient.send_task

source↗

send_task(
  agent_url: str,
  task: Task,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Block until a remote agent finishes processing a complete Task. This is the synchronous entry point for the same native or A2A submission path used by AgentClient.send_task().

Parameters

agent_urlstrrequired

Peer base URL or transport-specific URI that should receive the task.

taskTaskrequired

Caller-created task containing the request messages, metadata, and any continuation state.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Select native ProtoLink, require A2A, or automatically probe the native card before falling back to A2A on a 404 or 405 response.

Returns

taskTask

Processed remote task normalized into ProtoLink's task model.

Event-loop boundary

This method calls asyncio.run(). Do not invoke it from an active event loop; await client.send_task() instead.

SyncAgentClient.send_infer_task

methodprotolink.client.SyncAgentClient.send_infer_task

source↗

send_infer_task(
  query: str,
  agent_url: str,
  *,
  user: str | None = None,
  output_schema: dict[str, Any] | None = None,
  metadata: dict[str, Any] | None = None,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Synchronously build an inference task and submit it through the same native or A2A path as AgentClient.send_infer_task().

Parameters

querystrrequired

Prompt placed in the generated inference part.

agent_urlstrrequired

Peer address that should receive the generated task.

userstr | Nonedefault: None

Optional user identifier or context for the inference request.

output_schemadict[str, Any] | Nonedefault: None

Optional schema for the expected model output.

metadatadict[str, Any] | Nonedefault: None

Optional inference-part metadata.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Same native/A2A selection used by synchronous task submission.

Returns

taskTask

Processed remote inference task.

Event-loop boundary

This method calls asyncio.run(). In asynchronous code, await client.send_infer_task() instead.

SyncAgentClient.send_task_streaming

generator methodprotolink.client.SyncAgentClient.send_task_streaming

source↗

send_task_streaming(
  agent_url: str,
  task: Task,
) -> Iterator[Any]

Return a blocking iterator over the native streaming subscription. A daemon worker thread consumes the asynchronous iterator and forwards each decoded event, preserving the original event order and exception.

Parameters

agent_urlstrrequired

Peer address understood by the configured streaming transport.

taskTaskrequired

Task serialized into the streaming subscription request.

Yields

eventAny

Next transport-decoded status, progress, LLM, artifact, or completion event.

Raises

NotImplementedError

The configured transport does not support subscriptions.

stream error

The worker re-raises the original transport or remote exception in the consuming thread.

Streaming implementation

Unlike the other synchronous methods, this generator owns an asynchronous event loop in a daemon worker thread and relays values through a blocking queue.

SyncAgentClient.cancel_task

methodprotolink.client.SyncAgentClient.cancel_task

source↗

cancel_task(
  agent_url: str,
  task_id: str,
  *,
  reason: str | None = None,
  metadata: dict[str, Any] | None = None,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Task

Synchronously request best-effort cancellation of active remote work. Native cancellation sends the local task ID; A2A cancellation uses the server-assigned mapping retained by this client process.

Parameters

agent_urlstrrequired

Peer currently running the task.

task_idstrrequired

Local ProtoLink task identifier created by the caller.

reasonstr | Nonedefault: None

Optional human-readable cancellation reason propagated to the peer.

metadatadict[str, Any] | Nonedefault: None

Additional application context attached to the cancellation request.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Choose native cancellation, mapped A2A cancellation, or automatic selection.

Returns

taskTask

Remote task after the peer accepts and records the cancellation request.

Best effort

Acceptance does not roll back completed external side effects or forcibly interrupt synchronous work.

SyncAgentClient.compact_history

methodprotolink.client.SyncAgentClient.compact_history

source↗

compact_history(
  agent_url: str,
  *,
  strategy: HistoryCompactionStrategy = "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,
) -> HistoryCompactionResult

Block while the target agent reduces its LLM conversation history through the control plane. Supplying a persisted session loads, compacts, and saves that session; omitting it targets the agent LLM's current in-memory history.

Parameters

agent_urlstrrequired

Address of the target agent.

strategyLiteral["recent", "tokens", "summary"]default: "recent"

History-reduction algorithm selected by the server.

max_messagesintdefault: 20

Maximum message count used by recent-history compaction.

max_tokensintdefault: 4000

Estimated token budget used by token-aware compaction.

preserve_recentintdefault: 6

Number of newest messages protected from summarization.

summary_max_tokensintdefault: 512

Requested maximum token budget for the generated summary.

session_idstr | Nonedefault: None

Optional persisted conversation session to compact.

metadatadict[str, Any] | Nonedefault: None

Application context attached to the control request.

Returns

resultHistoryCompactionResult

Strategy metadata and the history's before-and-after message and token counts.

SyncAgentClient.describe_state

methodprotolink.client.SyncAgentClient.describe_state

source↗

describe_state(
  agent_url: str,
  *,
  session_id: str | None = None,
  stores: tuple[str, ...] | list[str] | None = None,
  include_data: bool = False,
  metadata: dict[str, Any] | None = None,
) -> StateOperationResult

Inspect selected persistent stores without mutating them. The blocking wrapper preserves the asynchronous method's idempotent request semantics and structured per-store report.

Parameters

agent_urlstrrequired

Address of the agent whose state should be inspected.

session_idstr | Nonedefault: None

Optional logical conversation session used to scope the inspection.

storestuple[str, ...] | list[str] | Nonedefault: None

Specific store names to inspect. Omission sends an empty selection so the server can apply its default scope.

include_databooldefault: False

Include stored values when supported instead of returning only availability and count metadata.

metadatadict[str, Any] | Nonedefault: None

Additional context sent with the control request.

Returns

resultStateOperationResult

Per-store enabled, missing, count, data, and error information.

SyncAgentClient.reset_state

methodprotolink.client.SyncAgentClient.reset_state

source↗

reset_state(
  agent_url: str,
  *,
  session_id: str | None = None,
  stores: tuple[str, ...] | list[str] | None = None,
  metadata: dict[str, Any] | None = None,
) -> StateOperationResult

Synchronously request deletion or reset of selected persistent state. A session ID narrows conversation deletion; without one, the server applies its configured default scope.

Parameters

agent_urlstrrequired

Address of the agent whose state should be reset.

session_idstr | Nonedefault: None

Optional conversation session to clear.

storestuple[str, ...] | list[str] | Nonedefault: None

Specific stores to reset. Omission leaves store selection to the server's default scope.

metadatadict[str, Any] | Nonedefault: None

Application context attached to the reset request.

Returns

resultStateOperationResult

Structured per-store reset, missing-store, and error report.

Destructive operation

The remote agent's policy and approval handler authorize the mutation. This client wrapper does not prompt for confirmation.

SyncAgentClient.compact_state

methodprotolink.client.SyncAgentClient.compact_state

source↗

compact_state(
  agent_url: str,
  *,
  session_id: str,
  strategy: HistoryCompactionStrategy = "tokens",
  max_messages: int = 20,
  max_tokens: int = 4000,
  preserve_recent: int = 6,
  summary_max_tokens: int = 512,
  metadata: dict[str, Any] | None = None,
) -> StateOperationResult

Compact one persisted conversation session and return its state report. This operation always targets the "conversation" store and therefore requires an explicit session identifier.

Parameters

agent_urlstrrequired

Address of the agent that owns the persisted conversation.

session_idstrrequired

Existing persisted session to compact.

strategyLiteral["recent", "tokens", "summary"]default: "tokens"

Conversation-history reduction strategy.

max_messagesintdefault: 20

Maximum message count used by recent-history compaction.

max_tokensintdefault: 4000

Estimated token budget used by token-aware compaction.

preserve_recentintdefault: 6

Number of newest messages protected from summarization.

summary_max_tokensintdefault: 512

Requested maximum token budget for the summary.

metadatadict[str, Any] | Nonedefault: None

Application context attached to the control request.

Returns

resultStateOperationResult

Compaction, missing-session, disabled-store, and error information.

SyncAgentClient.send_message

methodprotolink.client.SyncAgentClient.send_message

source↗

send_message(
  agent_url: str,
  message: Message,
  *,
  protocol: Literal["auto", "protolink", "a2a"] = "auto",
) -> Message

Wrap one Message in a new Task, wait for remote processing, and return the newest agent or assistant response. Use send_task() when the caller must retain artifacts, task state, context, or metadata.

Parameters

agent_urlstrrequired

Peer address that should receive the message.

messageMessagerequired

Input role and parts to place in the generated task.

protocolLiteral["auto", "protolink", "a2a"]default: "auto"

Use the same native, A2A, or automatic protocol selection as synchronous task submission.

Returns

messageMessage

Last returned message whose role is "agent" or "assistant".

Raises

RuntimeError

The returned task contains artifacts but no response message, or contains no response message at all.

SyncAgentClient.get_agent_card

methodprotolink.client.SyncAgentClient.get_agent_card

source↗

get_agent_card(
  agent_url: str,
) -> AgentCard

Fetch ProtoLink's native well-known card through the configured transport and parse it into a validated AgentCard.

Parameters

agent_urlstrrequired

Base address of the peer whose native ProtoLink card should be retrieved.

Returns

cardAgentCard

Validated identity, skills, interfaces, security, and capability metadata.

Example:

from protolink.client import AgentClient
from protolink.models import Task

client = AgentClient(transport="http", url="http://localhost:8000")

# No 'await' or 'async def' needed. Use the .sync property!
result = client.sync.send_infer_task("Hello, agent!", "http://localhost:8010")
print(result.get_last_part_content())

Synchronous streaming example:

client = AgentClient(transport="sse", url="http://localhost:8000")
task = Task.create_infer(prompt="Stream this response")

for event in client.sync.send_task_streaming("http://localhost:8010", task):
    print(event)

---

ClientRequestSpec

ClientRequestSpec defines the contract for an API endpoint in a transport-agnostic way.

It is the small description that sits between the high-level client and the wire transport. For example, “send a task” is a POST operation with a body and a Task response parser. HTTP turns that description into a route request, while WebSocket and gRPC place the same method and path in their envelopes. The client behavior stays identical because the specification describes the operation rather than the protocol.

Normal users rarely need to create request specs; the built-in Agent and Registry clients provide them. They become relevant when adding a new endpoint or implementing a custom client operation. At that point, idempotent deserves particular care because it authorizes the retry and duplicate-replay machinery.

frozen dataclassprotolink.models.ClientRequestSpec

source↗

ClientRequestSpec(
  name: str,
  path: str,
  method: HttpMethod,
  response_parser: Callable[[Any], Any] | None = None,
  request_source: RequestSourceType = "body",
  content_type: ContentType | None = None,
  accept: ContentType | None = None,
  channel: str = "default",
  idempotent: bool = False,
  headers: Mapping[str, str] | None = None,
)

Declare one transport-neutral operation. The dataclass is frozen so clients and transports can safely share a specification as a class-level constant.

Fields

namestrrequired

Stable operation name used in request contexts, metrics, and diagnostics.

pathstrrequired

Protocol-neutral endpoint path carried directly by HTTP or inside multiplexed envelopes.

methodHttpMethodrequired

Logical GET, POST, DELETE, PUT, or PATCH method used by routing and retry-method filtering.

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

Conversion from decoded wire data to a domain model.

request_sourceRequestSourceTypedefault: "body"

Body, query parameters, form, headers, path parameters, raw request, or no request data.

content_typeContentType | Nonedefault: None

Outbound media-type override.

acceptContentType | Nonedefault: None

Expected response media type.

channelstrdefault: "default"

Multiplexing lane; control operations use "control".

idempotentbooldefault: False

Explicit promise that replay under one idempotency key is safe. Retry machinery never retries a spec unless this is true.

headersMapping[str, str] | Nonedefault: None

Optional protocol headers; transports without a header concept may ignore them.

idempotent=True is an application-level safety promise, not an inference made from POST or a URL. Custom request specs should enable it only when repeating the operation with the same payload and idempotency key cannot apply the effect twice.

The channel field matters only to transports that multiplex several logical operations. Control requests such as cancellation use a separate channel so they are not forced to wait behind the long-running task they are intended to stop. Request/response transports may ignore the distinction while preserving the same client contract.

Built-in Request Specs

Spec	Request contract	Description
TASK_REQUEST	POST, /tasks/, channel: default, idempotent: True	Send a task to an agent. The task ID and idempotency key prevent duplicate execution.
TASK_CANCEL_REQUEST	POST, /tasks/cancel, channel: control, idempotent: True	Cancel an active task. Repeating cancellation has the same terminal effect.
COMPACT_HISTORY_REQUEST	POST, /llm/history/compact, channel: control, idempotent: False	Compact the target agent's LLM history.
DESCRIBE_STATE_REQUEST	POST, /state/describe, channel: control, idempotent: True	Inspect target agent state without mutating it.
RESET_STATE_REQUEST	POST, /state/reset, channel: control, idempotent: False	Reset target agent state.
COMPACT_STATE_REQUEST	POST, /state/compact, channel: control, idempotent: False	Compact target agent conversation state.
AGENT_CARD_REQUEST	GET, /.well-known/agent.json, channel: default, idempotent: True	Retrieve agent metadata.
TASK_STREAM_REQUEST	POST, /tasks/stream, channel: default, idempotent: False	Send a task and receive a live event stream. Streams are not replayed by the retry layer.

The outbound A2A adapter owns separate card-discovery and JSON-RPC request specifications. Card discovery is idempotent; SendMessage is deliberately not, so transport retry policy cannot create duplicate server-assigned tasks.

How It Works

When you call a method like send_task():

1. The client selects the appropriate ClientRequestSpec (for example, TASK_REQUEST).
2. It passes the specification and task data to transport.send().
3. The transport creates correlation and idempotency metadata, applies limits, and constructs the protocol-specific request.
4. The decoded response passes through response_parser, so the caller receives a Task, AgentCard, or another domain model rather than a raw wire dictionary.

If the spec is idempotent and the configured retry policy permits its method, the transport preserves one request ID and idempotency key across attempts. This pattern allows new endpoints without modifying transport implementations while keeping retry safety explicit at the operation boundary.