zhivex_ai.integrations.protocols

Public import paths and stability come from the installed artifact.

Use the root for Stable APIs and focused namespaces for extensions. Legacy root extension imports remain classified below.

A2AAgentCard

Beta · protocol

from zhivex_ai.integrations.protocols import A2AAgentCard

A2AAgentCard(name: 'str', description: 'str', url: 'str', version: 'str', skills: 'list[A2AAgentSkill]', protocol_version: 'str' = '1.0', preferred_transport: 'str' = 'HTTP+JSON', default_input_modes: 'list[str]' = <factory>, default_output_modes: 'list[str]' = <factory>, streaming: 'bool' = True, provider: 'dict[str, str] | None' = None, documentation_url: 'str | None' = None, security_schemes: 'dict[str, JsonValue]' = <factory>, security: 'list[dict[str, list[str]]]' = <factory>) -> None

A2AAgentCard(name: 'str', description: 'str', url: 'str', version: 'str', skills: 'list[A2AAgentSkill]', protocol_version: 'str' = '1.0', preferred_transport: 'str' = 'HTTP+JSON', default_input_modes: 'list[str]' = <factory>, default_output_modes: 'list[str]' = <factory>, streaming: 'bool' = True, provider: 'dict[str, str] | None' = None, documentation_url: 'str | None' = None, security_schemes: 'dict[str, JsonValue]' = <factory>, security: 'list[dict[str, list[str]]]' = <factory>)

A2AAgentExecutor

Beta · protocol

from zhivex_ai.integrations.protocols import A2AAgentExecutor

A2AAgentExecutor(agent: 'Agent', *, run_options_resolver: 'ProtocolRunOptionsResolver | None' = None, error_mapper: 'ProtocolErrorMapper | None' = None, on_protocol_event: 'ProtocolEventCallback | None' = None, limits: 'ProtocolLimits | None' = None) -> 'None'

In-process A2A v1 executor backed by a configured Zhivex Agent.

A2AAgentSkill

Beta · protocol

from zhivex_ai.integrations.protocols import A2AAgentSkill

A2AAgentSkill(id: 'str', name: 'str', description: 'str', tags: 'list[str]' = <factory>, examples: 'list[str]' = <factory>, input_modes: 'list[str] | None' = None, output_modes: 'list[str] | None' = None) -> None

A2AAgentSkill(id: 'str', name: 'str', description: 'str', tags: 'list[str]' = <factory>, examples: 'list[str]' = <factory>, input_modes: 'list[str] | None' = None, output_modes: 'list[str] | None' = None)

A2A_PROTOCOL_VERSION

Beta · protocol

from zhivex_ai.integrations.protocols import A2A_PROTOCOL_VERSION

A2A_PROTOCOL_VERSION (public type alias or constant; no callable signature)

AGUIEvent

Beta · protocol

from zhivex_ai.integrations.protocols import AGUIEvent

AGUIEvent(type: 'str', data: 'dict[str, Any]' = <factory>) -> None

AGUIEvent(type: 'str', data: 'dict[str, Any]' = <factory>)

HostedAgentRunOptions

Beta · protocol

from zhivex_ai.integrations.protocols import HostedAgentRunOptions

HostedAgentRunOptions(session: 'AgentSession | None' = None, deps: 'Any' = None, idempotency_key: 'str | None' = None, runtime: 'AgentRuntime | None' = None) -> None

Trusted, application-resolved options for one hosted agent run.

Authentication and tenant ownership remain application responsibilities.
In particular, ``deps`` is intentionally opaque and must never be persisted
by a protocol adapter.

ProtocolErrorMapper

Beta · protocol

from zhivex_ai.integrations.protocols import ProtocolErrorMapper

ProtocolErrorMapper(*args, **kwargs)

ProtocolEventCallback

Beta · protocol

from zhivex_ai.integrations.protocols import ProtocolEventCallback

ProtocolEventCallback(*args, **kwargs)

ProtocolInvocation

Beta · protocol

from zhivex_ai.integrations.protocols import ProtocolInvocation

ProtocolInvocation(protocol: 'str', action: 'str', agent_alias: 'str | None' = None, external_ids: 'dict[str, str]' = <factory>, request: 'Any' = None, payload: 'Mapping[str, Any] | None' = None) -> None

Trusted routing context passed only to application-owned resolvers/stores.

ProtocolLimits

Beta · protocol

from zhivex_ai.integrations.protocols import ProtocolLimits

ProtocolLimits(max_request_bytes: 'int' = 1048576, max_alias_chars: 'int' = 128, max_identifier_chars: 'int' = 256, max_messages: 'int' = 128, max_parts_per_message: 'int' = 64, max_text_chars: 'int' = 262144) -> None

Finite local parsing limits; not a rate-limit or tenancy policy.

ProtocolRunOptionsResolver

Beta · protocol

from zhivex_ai.integrations.protocols import ProtocolRunOptionsResolver

ProtocolRunOptionsResolver(*args, **kwargs)

create_a2a_agent_card

Beta · protocol

from zhivex_ai.integrations.protocols import create_a2a_agent_card

create_a2a_agent_card(agent: 'Agent', *, url: 'str', version: 'str', description: 'str | None' = None, skills: 'list[A2AAgentSkill] | None' = None, provider: 'dict[str, str] | None' = None, documentation_url: 'str | None' = None, security_schemes: 'dict[str, JsonValue] | None' = None, security: 'list[dict[str, list[str]]] | None' = None) -> 'A2AAgentCard'

create_a2a_app

Beta · protocol

from zhivex_ai.integrations.protocols import create_a2a_app

create_a2a_app(*, executor: 'A2AAgentExecutor', card: 'A2AAgentCard', authorize: 'Callable[[Any], bool | Awaitable[bool]] | None' = None, max_request_bytes: 'int | None' = None, limits: 'ProtocolLimits | None' = None, task_store: 'Any' = None, request_context_builder: 'Any' = None, queue_manager: 'Any' = None)

Create an A2A v1 server using the official Python SDK.

The app exposes both HTTP+JSON under ``/a2a`` and JSON-RPC at
``/a2a/rpc``. The official task store and request handler own the wire
protocol, task lifecycle, version headers, and streaming envelopes.

stream_agent_ag_ui

Beta · protocol

from zhivex_ai.integrations.protocols import stream_agent_ag_ui

stream_agent_ag_ui(*, agent: 'Agent', prompt: 'str', thread_id: 'str', run_id: 'str | None' = None, run_options: 'HostedAgentRunOptions | None' = None, run_options_resolver: 'ProtocolRunOptionsResolver | None' = None, error_mapper: 'ProtocolErrorMapper | None' = None, on_protocol_event: 'ProtocolEventCallback | None' = None, limits: 'ProtocolLimits | None' = None) -> 'AsyncIterable[dict[str, Any]]'

Translate a Zhivex agent stream into canonical AG-UI lifecycle events.

to_ag_ui_sse_response

Beta · protocol

from zhivex_ai.integrations.protocols import to_ag_ui_sse_response

to_ag_ui_sse_response(source: 'AsyncIterable[dict[str, Any]]') -> 'HTTPResponse'

Encode AG-UI events with the official protocol encoder.