Saltar a contenido

API reference

Generated from the apogee-ai-comunication source with mkdocstrings. Every symbol below is exported from apogee_ai_comunication, so it is part of the supported public surface.

Other

AgentCard dataclass

Python
AgentCard(name: str, version: str = '0.1.0', description: str = '', url: str | None = None, protocols: tuple[str, ...] = (), capabilities: tuple[AgentSkill, ...] = (), auth_schemes: tuple[str, ...] = ('none',), extra: dict[str, object] = dict())

Public metadata used for cross-protocol agent discovery.

name instance-attribute

Python
name: str

version class-attribute instance-attribute

Python
version: str = '0.1.0'

description class-attribute instance-attribute

Python
description: str = ''

url class-attribute instance-attribute

Python
url: str | None = None

protocols class-attribute instance-attribute

Python
protocols: tuple[str, ...] = ()

capabilities class-attribute instance-attribute

Python
capabilities: tuple[AgentSkill, ...] = ()

auth_schemes class-attribute instance-attribute

Python
auth_schemes: tuple[str, ...] = ('none',)

extra class-attribute instance-attribute

Python
extra: dict[str, object] = field(default_factory=dict)

AgentCardMapper

to_dto staticmethod

Python
to_dto(card: AgentCard) -> AgentCardDTO
Source code in apogee_ai_comunication/application/mappers/agent_card_mapper.py
Python
@staticmethod
def to_dto(card: AgentCard) -> AgentCardDTO:
    return AgentCardDTO(
        name=card.name,
        version=card.version,
        description=card.description,
        url=card.url,
        protocols=list(card.protocols),
        capabilities=[
            AgentSkillDTO(
                name=s.name,
                description=s.description,
                input_modes=list(s.input_modes),
                output_modes=list(s.output_modes),
                tags=list(s.tags),
            )
            for s in card.capabilities
        ],
        auth_schemes=list(card.auth_schemes),
    )

from_dto staticmethod

Python
from_dto(dto: AgentCardDTO) -> AgentCard
Source code in apogee_ai_comunication/application/mappers/agent_card_mapper.py
Python
@staticmethod
def from_dto(dto: AgentCardDTO) -> AgentCard:
    return AgentCard(
        name=dto.name,
        version=dto.version,
        description=dto.description,
        url=dto.url,
        protocols=tuple(dto.protocols),
        capabilities=tuple(
            AgentSkill(
                name=s.name,
                description=s.description,
                input_modes=tuple(s.input_modes),
                output_modes=tuple(s.output_modes),
                tags=tuple(s.tags),
            )
            for s in dto.capabilities
        ),
        auth_schemes=tuple(dto.auth_schemes),
    )

AgentMessage dataclass

Python
AgentMessage(role: MessageRole, parts: tuple[MessagePart, ...] = (), name: str | None = None, metadata: dict[str, object] = dict())

Neutral chat-style message that any protocol can map onto its wire-format.

role instance-attribute

Python
role: MessageRole

parts class-attribute instance-attribute

Python
parts: tuple[MessagePart, ...] = ()

name class-attribute instance-attribute

Python
name: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, object] = field(default_factory=dict)

AgentMessageMapper

to_dto staticmethod

Python
to_dto(msg: AgentMessage) -> AgentMessageDTO
Source code in apogee_ai_comunication/application/mappers/agent_message_mapper.py
Python
@staticmethod
def to_dto(msg: AgentMessage) -> AgentMessageDTO:
    return AgentMessageDTO(
        role=msg.role.value,
        parts=[
            MessagePartDTO(kind=p.kind, text=p.text, data=p.data, mime_type=p.mime_type)
            for p in msg.parts
        ],
        name=msg.name,
        metadata=dict(msg.metadata),
    )

from_dto staticmethod

Python
from_dto(dto: AgentMessageDTO) -> AgentMessage
Source code in apogee_ai_comunication/application/mappers/agent_message_mapper.py
Python
@staticmethod
def from_dto(dto: AgentMessageDTO) -> AgentMessage:
    return AgentMessage(
        role=MessageRole(dto.role),
        parts=tuple(
            MessagePart(kind=p.kind, text=p.text, data=p.data, mime_type=p.mime_type)
            for p in dto.parts
        ),
        name=dto.name,
        metadata=dict(dto.metadata),
    )

AgentSkill dataclass

Python
AgentSkill(name: str, description: str = '', input_modes: tuple[str, ...] = ('text',), output_modes: tuple[str, ...] = ('text',), tags: tuple[str, ...] = ())

A discrete capability advertised by the agent.

name instance-attribute

Python
name: str

description class-attribute instance-attribute

Python
description: str = ''

input_modes class-attribute instance-attribute

Python
input_modes: tuple[str, ...] = ('text',)

output_modes class-attribute instance-attribute

Python
output_modes: tuple[str, ...] = ('text',)

tags class-attribute instance-attribute

Python
tags: tuple[str, ...] = ()

AgentTask dataclass

Python
AgentTask(task_id: str, state: TaskState = SUBMITTED, history: tuple[AgentMessage, ...] = (), artifacts: tuple[TaskArtifact, ...] = (), metadata: dict[str, object] = dict())

A long-running unit of work, observable via SSE.

task_id instance-attribute

Python
task_id: str

state class-attribute instance-attribute

Python
state: TaskState = SUBMITTED

history class-attribute instance-attribute

Python
history: tuple[AgentMessage, ...] = ()

artifacts class-attribute instance-attribute

Python
artifacts: tuple[TaskArtifact, ...] = ()

metadata class-attribute instance-attribute

Python
metadata: dict[str, object] = field(default_factory=dict)

AgentTaskMapper

to_dto staticmethod

Python
to_dto(task: AgentTask) -> AgentTaskDTO
Source code in apogee_ai_comunication/application/mappers/agent_task_mapper.py
Python
@staticmethod
def to_dto(task: AgentTask) -> AgentTaskDTO:
    return AgentTaskDTO(
        task_id=task.task_id,
        state=task.state.value,
        history=[AgentMessageMapper.to_dto(m) for m in task.history],
        artifacts=[
            TaskArtifactDTO(
                name=a.name, mime_type=a.mime_type, data=dict(a.data), url=a.url
            )
            for a in task.artifacts
        ],
        metadata=dict(task.metadata),
    )

from_dto staticmethod

Python
from_dto(dto: AgentTaskDTO) -> AgentTask
Source code in apogee_ai_comunication/application/mappers/agent_task_mapper.py
Python
@staticmethod
def from_dto(dto: AgentTaskDTO) -> AgentTask:
    return AgentTask(
        task_id=dto.task_id,
        state=TaskState(dto.state),
        history=tuple(AgentMessageMapper.from_dto(m) for m in dto.history),
        artifacts=tuple(
            TaskArtifact(
                name=a.name, mime_type=a.mime_type, data=dict(a.data), url=a.url
            )
            for a in dto.artifacts
        ),
        metadata=dict(dto.metadata),
    )

AiCommFactory

Convenience factory keyed by ProtocolKind.

engine="native" (default) picks zero-dep adapters; any other value attempts to load the optional engine and may raise ProviderNotInstalledError.

build_server staticmethod

Python
build_server(kind: ProtocolKind, config: ServerConfig, *, engine: str = 'native', agent_card: AgentCard | None = None, **kwargs: Any) -> Any
Source code in apogee_ai_comunication/infrastructure/factory/ai_comm_factory.py
Python
@staticmethod
def build_server(
    kind: ProtocolKind,
    config: ServerConfig,
    *,
    engine: str = "native",
    agent_card: AgentCard | None = None,
    **kwargs: Any,
) -> Any:
    if kind is ProtocolKind.MCP:
        if engine == "native":
            from apogee_ai_comunication.infrastructure.mcp.native_mcp_server import (
                NativeMcpServer,
            )

            return NativeMcpServer(config)
        if engine == "fastmcp":
            from apogee_ai_comunication.infrastructure.mcp.fastmcp_server import (
                FastMcpServer,
            )

            return FastMcpServer(config)
    if kind is ProtocolKind.A2A:
        card = agent_card or AgentCard(name="apogee-a2a-agent")
        if engine == "native":
            from apogee_ai_comunication.infrastructure.a2a.native_a2a_server import (
                NativeA2aServer,
            )

            return NativeA2aServer(config, card, kwargs.get("worker"))
        if engine == "a2a-sdk":
            from apogee_ai_comunication.infrastructure.a2a.a2a_sdk_server import (
                A2aSdkServer,
            )

            return A2aSdkServer(config, card)
    if kind is ProtocolKind.AG_UI:
        if engine == "native":
            from apogee_ai_comunication.infrastructure.ag_ui.native_ag_ui_server import (
                NativeAgUiServer,
            )

            return NativeAgUiServer(config, kwargs.get("handler"))
        if engine == "ag-ui-protocol":
            from apogee_ai_comunication.infrastructure.ag_ui.ag_ui_protocol_server import (
                AgUiProtocolServer,
            )

            return AgUiProtocolServer(config)
    if kind is ProtocolKind.ACP:
        if engine == "native":
            from apogee_ai_comunication.infrastructure.acp.native_acp_server import (
                NativeAcpServer,
            )

            return NativeAcpServer(config, kwargs.get("handler"))
        if engine == "acp-sdk":
            from apogee_ai_comunication.infrastructure.acp.acp_sdk_server import (
                AcpSdkServer,
            )

            return AcpSdkServer(config)
    if kind is ProtocolKind.MCP_UI:
        from apogee_ai_comunication.infrastructure.mcp_ui.native_ui_provider import (
            NativeUiProvider,
        )

        return NativeUiProvider()
    if kind is ProtocolKind.A2UI:
        from apogee_ai_comunication.infrastructure.a2ui.native_renderer import (
            NativeA2uiRenderer,
        )

        return NativeA2uiRenderer()
    raise ProtocolError(f"unsupported (kind={kind!s}, engine={engine})")

build_client staticmethod

Python
build_client(kind: ProtocolKind, config: ClientConfig, *, engine: str = 'native', server: Any = None) -> Any
Source code in apogee_ai_comunication/infrastructure/factory/ai_comm_factory.py
Python
@staticmethod
def build_client(
    kind: ProtocolKind,
    config: ClientConfig,
    *,
    engine: str = "native",
    server: Any = None,
) -> Any:
    if kind is ProtocolKind.MCP:
        if engine == "native":
            from apogee_ai_comunication.infrastructure.mcp.native_mcp_client import (
                NativeMcpClient,
            )

            return NativeMcpClient(config, dispatcher=getattr(server, "dispatch", None))
        if engine == "fastmcp":
            from apogee_ai_comunication.infrastructure.mcp.fastmcp_client import (
                FastMcpClient,
            )

            return FastMcpClient(config)
    if kind is ProtocolKind.A2A:
        from apogee_ai_comunication.infrastructure.a2a.native_a2a_client import (
            NativeA2aClient,
        )

        if server is None:
            raise ProtocolError("A2A native client requires a server reference")
        return NativeA2aClient(config, server)
    if kind is ProtocolKind.ACP:
        from apogee_ai_comunication.infrastructure.acp.native_acp_client import (
            NativeAcpClient,
        )

        if server is None:
            raise ProtocolError("ACP native client requires a server reference")
        return NativeAcpClient(config, server)
    raise ProtocolError(f"unsupported client (kind={kind!s}, engine={engine})")

AuthConfig dataclass

Python
AuthConfig(scheme: AuthScheme = NONE, token: str | None = None, issuer: str | None = None, audience: str | None = None)

scheme class-attribute instance-attribute

Python
scheme: AuthScheme = NONE

token class-attribute instance-attribute

Python
token: str | None = None

issuer class-attribute instance-attribute

Python
issuer: str | None = None

audience class-attribute instance-attribute

Python
audience: str | None = None

AuthScheme

Bases: str, Enum

NONE class-attribute instance-attribute

Python
NONE = 'none'

BEARER class-attribute instance-attribute

Python
BEARER = 'bearer'

OAUTH2 class-attribute instance-attribute

Python
OAUTH2 = 'oauth2'

MTLS class-attribute instance-attribute

Python
MTLS = 'mtls'

ClientConfig dataclass

Python
ClientConfig(base_url: str, auth_scheme: AuthScheme = NONE, auth_token: str | None = None, timeout_seconds: float = 30.0, extras: dict[str, object] = dict())

base_url instance-attribute

Python
base_url: str

auth_scheme class-attribute instance-attribute

Python
auth_scheme: AuthScheme = NONE

auth_token class-attribute instance-attribute

Python
auth_token: str | None = None

timeout_seconds class-attribute instance-attribute

Python
timeout_seconds: float = 30.0

extras class-attribute instance-attribute

Python
extras: dict[str, object] = field(default_factory=dict)

ComponentCatalog dataclass

Python
ComponentCatalog(components: dict[str, ComponentSpec] = dict())

components class-attribute instance-attribute

Python
components: dict[str, ComponentSpec] = field(default_factory=dict)

register

Python
register(spec: ComponentSpec) -> 'ComponentCatalog'
Source code in apogee_ai_comunication/infrastructure/a2ui/component_catalog.py
Python
def register(self, spec: ComponentSpec) -> "ComponentCatalog":
    new = dict(self.components)
    new[spec.type] = spec
    return ComponentCatalog(components=new)

is_known

Python
is_known(type_: str) -> bool
Source code in apogee_ai_comunication/infrastructure/a2ui/component_catalog.py
Python
def is_known(self, type_: str) -> bool:
    return type_ in self.components

spec_for

Python
spec_for(type_: str) -> ComponentSpec | None
Source code in apogee_ai_comunication/infrastructure/a2ui/component_catalog.py
Python
def spec_for(self, type_: str) -> ComponentSpec | None:
    return self.components.get(type_)

DEFAULT_CATALOG module-attribute

Python
DEFAULT_CATALOG = ComponentCatalog(components={(type): s for s in _DEFAULT_SPECS})

IframeDescriptor dataclass

Python
IframeDescriptor(src: str, sandbox: tuple[str, ...] = ('allow-scripts', 'allow-forms', 'allow-same-origin'), width: str = '100%', height: str = '600', extra_attrs: dict[str, str] = dict())

src instance-attribute

Python
src: str

sandbox class-attribute instance-attribute

Python
sandbox: tuple[str, ...] = ('allow-scripts', 'allow-forms', 'allow-same-origin')

width class-attribute instance-attribute

Python
width: str = '100%'

height class-attribute instance-attribute

Python
height: str = '600'

extra_attrs class-attribute instance-attribute

Python
extra_attrs: dict[str, str] = field(default_factory=dict)

to_html

Python
to_html() -> str
Source code in apogee_ai_comunication/infrastructure/mcp_ui/iframe_descriptor.py
Python
def to_html(self) -> str:
    attrs = " ".join(f'{k}="{v}"' for k, v in self.extra_attrs.items())
    return (
        f'<iframe src="{self.src}" '
        f'sandbox="{" ".join(self.sandbox)}" '
        f'width="{self.width}" height="{self.height}" {attrs}></iframe>'
    )

InMemoryObservabilityEmitter

Python
InMemoryObservabilityEmitter()

Bases: IObservabilityEmitter

Source code in apogee_ai_comunication/infrastructure/observability/in_memory_observability_emitter.py
Python
def __init__(self) -> None:
    self.events: list[tuple[str, dict[str, object]]] = []

events instance-attribute

Python
events: list[tuple[str, dict[str, object]]] = []

emit

Python
emit(name: str, payload: dict[str, object]) -> None
Source code in apogee_ai_comunication/infrastructure/observability/in_memory_observability_emitter.py
Python
def emit(self, name: str, payload: dict[str, object]) -> None:
    self.events.append((name, dict(payload)))

clear

Python
clear() -> None
Source code in apogee_ai_comunication/infrastructure/observability/in_memory_observability_emitter.py
Python
def clear(self) -> None:
    self.events.clear()

InMemoryToolRegistry

Python
InMemoryToolRegistry()

Bases: IToolRegistry

Thread-safe-by-coroutine, in-memory registry of tools.

Source code in apogee_ai_comunication/infrastructure/tool_registry/in_memory_tool_registry.py
Python
def __init__(self) -> None:
    self._defs: dict[str, ToolDefinition] = {}
    self._handlers: dict[str, ToolHandler] = {}

register

Python
register(definition: ToolDefinition, handler: ToolHandler) -> None
Source code in apogee_ai_comunication/infrastructure/tool_registry/in_memory_tool_registry.py
Python
def register(self, definition: ToolDefinition, handler: ToolHandler) -> None:
    self._defs[definition.name] = definition
    self._handlers[definition.name] = handler

list_tools

Python
list_tools() -> list[ToolDefinition]
Source code in apogee_ai_comunication/infrastructure/tool_registry/in_memory_tool_registry.py
Python
def list_tools(self) -> list[ToolDefinition]:
    return list(self._defs.values())

get_handler

Python
get_handler(name: str) -> ToolHandler | None
Source code in apogee_ai_comunication/infrastructure/tool_registry/in_memory_tool_registry.py
Python
def get_handler(self, name: str) -> ToolHandler | None:
    return self._handlers.get(name)

invoke async

Python
invoke(invocation: ToolInvocation) -> ToolResult
Source code in apogee_ai_comunication/infrastructure/tool_registry/in_memory_tool_registry.py
Python
async def invoke(self, invocation: ToolInvocation) -> ToolResult:
    handler = self._handlers.get(invocation.name)
    if handler is None:
        raise ToolInvocationError(
            f"Unknown tool: {invocation.name}", tool_name=invocation.name
        )
    try:
        return await handler(invocation)
    except ToolInvocationError:
        raise
    except Exception as exc:  # noqa: BLE001
        return ToolResult(
            call_id=invocation.call_id,
            content=None,
            is_error=True,
            error_message=str(exc),
        )

LifecycleEvents

text_delta staticmethod

Python
text_delta(text: str, *, sequence: int | None = None) -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
@staticmethod
def text_delta(text: str, *, sequence: int | None = None) -> UiEvent:
    return UiEvent(
        kind=UiEventKind.TEXT_DELTA, payload={"delta": text}, sequence=sequence
    )

tool_call staticmethod

Python
tool_call(name: str, arguments: dict[str, object]) -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
@staticmethod
def tool_call(name: str, arguments: dict[str, object]) -> UiEvent:
    return UiEvent(
        kind=UiEventKind.TOOL_CALL,
        payload={"name": name, "arguments": dict(arguments)},
    )

tool_result staticmethod

Python
tool_result(name: str, content: object, *, is_error: bool = False) -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
@staticmethod
def tool_result(name: str, content: object, *, is_error: bool = False) -> UiEvent:
    return UiEvent(
        kind=UiEventKind.TOOL_RESULT,
        payload={"name": name, "content": content, "is_error": is_error},
    )

state_patch staticmethod

Python
state_patch(patch: dict[str, object]) -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
@staticmethod
def state_patch(patch: dict[str, object]) -> UiEvent:
    return UiEvent(kind=UiEventKind.STATE_PATCH, payload={"patch": dict(patch)})

error staticmethod

Python
error(message: str) -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
@staticmethod
def error(message: str) -> UiEvent:
    return UiEvent(kind=UiEventKind.ERROR, payload={"message": message})

MessagePart dataclass

Python
MessagePart(kind: str, text: str | None = None, data: dict[str, object] | None = None, mime_type: str | None = None)

A piece of message content. kind ∈ text|tool_call|tool_result|file|ui.

kind instance-attribute

Python
kind: str

text class-attribute instance-attribute

Python
text: str | None = None

data class-attribute instance-attribute

Python
data: dict[str, object] | None = None

mime_type class-attribute instance-attribute

Python
mime_type: str | None = None

MessageRole

Bases: str, Enum

USER class-attribute instance-attribute

Python
USER = 'user'

ASSISTANT class-attribute instance-attribute

Python
ASSISTANT = 'assistant'

SYSTEM class-attribute instance-attribute

Python
SYSTEM = 'system'

TOOL class-attribute instance-attribute

Python
TOOL = 'tool'

NativeA2aClient

Python
NativeA2aClient(config: ClientConfig, server: NativeA2aServer)

Bases: IA2aClient

Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_client.py
Python
def __init__(self, config: ClientConfig, server: NativeA2aServer) -> None:
    self._config = config
    self._server = server

discover async

Python
discover() -> AgentCard
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_client.py
Python
async def discover(self) -> AgentCard:
    return self._server.agent_card

send_task async

Python
send_task(message: AgentMessage) -> AgentTask
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_client.py
Python
async def send_task(self, message: AgentMessage) -> AgentTask:
    return await self._server.handle_task(message)

watch_task async

Python
watch_task(task_id: str) -> AsyncIterator[AgentTask]
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_client.py
Python
async def watch_task(self, task_id: str) -> AsyncIterator[AgentTask]:
    async for update in self._server.stream_task(task_id):
        yield update

close async

Python
close() -> None
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_client.py
Python
async def close(self) -> None:
    return None

NativeA2aServer

Python
NativeA2aServer(config: ServerConfig, agent_card: AgentCard, worker: TaskWorker | None = None)

Bases: IA2aServer

In-process A2A server suitable for native deployments and tests.

Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_server.py
Python
def __init__(
    self,
    config: ServerConfig,
    agent_card: AgentCard,
    worker: TaskWorker | None = None,
) -> None:
    self._config = config
    self._card = agent_card
    self._worker: TaskWorker = worker or self._default_worker
    self._tasks: dict[str, AgentTask] = {}
    self._streams: dict[str, asyncio.Queue[AgentTask]] = {}
    self._stop_event = asyncio.Event()

agent_card property

Python
agent_card: AgentCard

handle_task async

Python
handle_task(message: AgentMessage) -> AgentTask
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_server.py
Python
async def handle_task(self, message: AgentMessage) -> AgentTask:
    task_id = uuid.uuid4().hex
    task = AgentTask(task_id=task_id, state=TaskState.SUBMITTED, history=(message,))
    self._tasks[task_id] = task
    queue: asyncio.Queue[AgentTask] = asyncio.Queue()
    self._streams[task_id] = queue
    await queue.put(task)
    asyncio.create_task(self._run_task(task, message, queue))
    return task

stream_task async

Python
stream_task(task_id: str) -> AsyncIterator[AgentTask]
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_server.py
Python
async def stream_task(self, task_id: str) -> AsyncIterator[AgentTask]:
    queue = self._streams.get(task_id)
    if queue is None:
        raise KeyError(f"unknown task: {task_id}")
    while True:
        update = await queue.get()
        if update.metadata.get("_sentinel"):
            return
        yield update

run async

Python
run() -> None
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_server.py
Python
async def run(self) -> None:
    await self._stop_event.wait()

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/infrastructure/a2a/native_a2a_server.py
Python
async def stop(self) -> None:
    self._stop_event.set()

NativeA2uiRenderer

Python
NativeA2uiRenderer(catalog: ComponentCatalog | None = None)

Bases: IA2uiRenderer

Source code in apogee_ai_comunication/infrastructure/a2ui/native_renderer.py
Python
def __init__(self, catalog: ComponentCatalog | None = None) -> None:
    self._catalog = catalog or DEFAULT_CATALOG

validate

Python
validate(root: UiComponent) -> list[str]
Source code in apogee_ai_comunication/infrastructure/a2ui/native_renderer.py
Python
def validate(self, root: UiComponent) -> list[str]:
    problems: list[str] = []
    self._walk(root, problems)
    return problems

render

Python
render(root: UiComponent) -> dict[str, object]
Source code in apogee_ai_comunication/infrastructure/a2ui/native_renderer.py
Python
def render(self, root: UiComponent) -> dict[str, object]:
    problems = self.validate(root)
    if problems:
        return {"valid": False, "problems": problems}
    return {"valid": True, "tree": self._serialize(root)}

NativeAcpClient

Python
NativeAcpClient(config: ClientConfig, server: NativeAcpServer)

Bases: IAcpClient

Source code in apogee_ai_comunication/infrastructure/acp/native_acp_client.py
Python
def __init__(self, config: ClientConfig, server: NativeAcpServer) -> None:
    self._config = config
    self._server = server

send async

Python
send(message: AgentMessage) -> AgentMessage
Source code in apogee_ai_comunication/infrastructure/acp/native_acp_client.py
Python
async def send(self, message: AgentMessage) -> AgentMessage:
    return await self._server.receive(message)

close async

Python
close() -> None
Source code in apogee_ai_comunication/infrastructure/acp/native_acp_client.py
Python
async def close(self) -> None:
    return None

NativeAcpServer

Python
NativeAcpServer(config: ServerConfig, handler: AcpHandler | None = None)

Bases: IAcpServer

Minimal in-process ACP server.

Source code in apogee_ai_comunication/infrastructure/acp/native_acp_server.py
Python
def __init__(self, config: ServerConfig, handler: AcpHandler | None = None) -> None:
    self._config = config
    self._handler = handler or self._default_handler
    self._stop_event = asyncio.Event()

receive async

Python
receive(message: AgentMessage) -> AgentMessage
Source code in apogee_ai_comunication/infrastructure/acp/native_acp_server.py
Python
async def receive(self, message: AgentMessage) -> AgentMessage:
    return await self._handler(message)

run async

Python
run() -> None
Source code in apogee_ai_comunication/infrastructure/acp/native_acp_server.py
Python
async def run(self) -> None:
    await self._stop_event.wait()

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/infrastructure/acp/native_acp_server.py
Python
async def stop(self) -> None:
    self._stop_event.set()

NativeAgUiServer

Python
NativeAgUiServer(config: ServerConfig, handler: AgUiHandler | None = None)

Bases: IAgUiServer

Pumps events through a NativeEventEmitter for an in-process AG-UI session.

Source code in apogee_ai_comunication/infrastructure/ag_ui/native_ag_ui_server.py
Python
def __init__(self, config: ServerConfig, handler: AgUiHandler | None = None) -> None:
    self._config = config
    self._handler = handler or self._default_handler

stream_events async

Python
stream_events(message: AgentMessage) -> AsyncIterator[UiEvent]
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_ag_ui_server.py
Python
async def stream_events(self, message: AgentMessage) -> AsyncIterator[UiEvent]:
    emitter = NativeEventEmitter()
    run_id = "run-1"
    await emitter.emit(start_event(run_id))

    async def runner() -> None:
        try:
            await self._handler(message, emitter)
        finally:
            await emitter.emit(finish_event(run_id))
            await emitter.close()

    import asyncio

    asyncio.create_task(runner())
    async for event in emitter.stream():
        yield event

run async

Python
run() -> None
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_ag_ui_server.py
Python
async def run(self) -> None:
    return None

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_ag_ui_server.py
Python
async def stop(self) -> None:
    return None

NativeEventEmitter

Python
NativeEventEmitter()

Bases: IAgUiEventEmitter

Buffered event queue → SSE / DTO stream.

Source code in apogee_ai_comunication/infrastructure/ag_ui/native_event_emitter.py
Python
def __init__(self) -> None:
    self._queue: asyncio.Queue[UiEvent | None] = asyncio.Queue()
    self._closed = False

emit async

Python
emit(event: UiEvent) -> None
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_event_emitter.py
Python
async def emit(self, event: UiEvent) -> None:
    if self._closed:
        return
    await self._queue.put(event)

close async

Python
close() -> None
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_event_emitter.py
Python
async def close(self) -> None:
    if not self._closed:
        self._closed = True
        await self._queue.put(None)

stream async

Python
stream() -> AsyncIterator[UiEvent]
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_event_emitter.py
Python
async def stream(self) -> AsyncIterator[UiEvent]:
    while True:
        item = await self._queue.get()
        if item is None:
            return
        yield item

stream_sse async

Python
stream_sse() -> AsyncIterator[str]
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_event_emitter.py
Python
async def stream_sse(self) -> AsyncIterator[str]:
    seq = 0
    async for event in self.stream():
        seq += 1
        yield SseEncoder.encode(
            UiEventMapper.to_dto(event).model_dump(),
            event=event.kind.value,
            id_=str(seq),
        )

stream_dto async

Python
stream_dto() -> AsyncIterator[UiEventDTO]
Source code in apogee_ai_comunication/infrastructure/ag_ui/native_event_emitter.py
Python
async def stream_dto(self) -> AsyncIterator[UiEventDTO]:
    async for event in self.stream():
        yield UiEventMapper.to_dto(event)

NativeMcpClient

Python
NativeMcpClient(config: ClientConfig, *, pipe: InMemoryPipe | None = None, dispatcher=None, http_client: AsyncClient | None = None)

Bases: IMcpClient

Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_client.py
Python
def __init__(
    self,
    config: ClientConfig,
    *,
    pipe: InMemoryPipe | None = None,
    dispatcher=None,
    http_client: httpx.AsyncClient | None = None,
) -> None:
    self._config = config
    self._pipe = pipe
    self._dispatcher = dispatcher  # callable (str) -> awaitable[str|None]
    self._http = http_client
    self._owns_http = http_client is None
    self._counter = 0

list_tools async

Python
list_tools() -> list[ToolDefinition]
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_client.py
Python
async def list_tools(self) -> list[ToolDefinition]:
    result = await self._request("tools/list")
    return [
        ToolDefinition(
            name=t["name"],
            description=t.get("description", ""),
            input_schema=dict(t.get("inputSchema", {})),
        )
        for t in result.get("tools", [])
    ]

call_tool async

Python
call_tool(invocation: ToolInvocation) -> ToolResult
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_client.py
Python
async def call_tool(self, invocation: ToolInvocation) -> ToolResult:
    result = await self._request(
        "tools/call",
        {
            "name": invocation.name,
            "arguments": dict(invocation.arguments),
            "call_id": invocation.call_id,
        },
    )
    is_error = bool(result.get("isError"))
    content_blocks = result.get("content") or []
    first = content_blocks[0] if content_blocks else {}
    text = first.get("text") if isinstance(first, dict) else None
    return ToolResult(
        call_id=invocation.call_id,
        content=text,
        is_error=is_error,
        error_message=text if is_error else None,
    )

get_resource async

Python
get_resource(uri: str) -> ResourceDescriptor
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_client.py
Python
async def get_resource(self, uri: str) -> ResourceDescriptor:
    result = await self._request("resources/read", {"uri": uri})
    contents = result.get("contents") or []
    if not contents:
        raise ProtocolError(f"empty resource: {uri}", code=-32602)
    first = contents[0]
    return ResourceDescriptor(
        uri=first.get("uri", uri),
        name=first.get("uri", uri),
        mime_type=first.get("mimeType", "text/plain"),
        contents=first.get("text"),
    )

close async

Python
close() -> None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_client.py
Python
async def close(self) -> None:
    if self._http is not None and self._owns_http:
        await self._http.aclose()

NativeMcpServer

Python
NativeMcpServer(config: ServerConfig)

Bases: IMcpServer

Pure-Python MCP server with the methods required by the public spec.

Implements

initialize, tools/list, tools/call, resources/list, resources/read, prompts/list, prompts/get, ping.

Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
def __init__(self, config: ServerConfig) -> None:
    self._config = config
    self._registry = InMemoryToolRegistry()
    self._resources: dict[str, ResourceDescriptor] = {}
    self._prompts: dict[str, PromptTemplate] = {}
    self._stop_event = asyncio.Event()

PROTOCOL_VERSION class-attribute instance-attribute

Python
PROTOCOL_VERSION = '2024-11-05'

registry property

Python
registry: InMemoryToolRegistry

resources property

Python
resources: dict[str, ResourceDescriptor]

prompts property

Python
prompts: dict[str, PromptTemplate]

register_tool

Python
register_tool(definition: ToolDefinition, handler: ToolHandler) -> None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
def register_tool(self, definition: ToolDefinition, handler: ToolHandler) -> None:
    self._registry.register(definition, handler)

register_resource

Python
register_resource(resource: ResourceDescriptor) -> None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
def register_resource(self, resource: ResourceDescriptor) -> None:
    self._resources[resource.uri] = resource

register_prompt

Python
register_prompt(prompt: PromptTemplate) -> None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
def register_prompt(self, prompt: PromptTemplate) -> None:
    self._prompts[prompt.name] = prompt

run async

Python
run() -> None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
async def run(self) -> None:
    await self._stop_event.wait()

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
async def stop(self) -> None:
    self._stop_event.set()

tool

Python
tool(definition: ToolDefinition) -> Callable[[Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]]

Decorator: registers a coroutine as the handler for definition.

Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
def tool(self, definition: ToolDefinition) -> Callable[
    [Callable[..., Awaitable[Any]]], Callable[..., Awaitable[Any]]
]:
    """Decorator: registers a coroutine as the handler for `definition`."""

    def decorator(
        fn: Callable[..., Awaitable[Any]],
    ) -> Callable[..., Awaitable[Any]]:
        async def handler(invocation: ToolInvocation):
            from apogee_ai_comunication.domain.entities.tool_result import ToolResult

            content = await fn(**dict(invocation.arguments))
            return ToolResult(call_id=invocation.call_id, content=content)

        self._registry.register(definition, handler)
        return fn

    return decorator

dispatch async

Python
dispatch(raw: str | bytes) -> str | None
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
async def dispatch(self, raw: str | bytes) -> str | None:
    envelope = JsonRpcCodec.decode(raw)
    method = envelope.get("method")
    params = envelope.get("params") or {}
    req_id = envelope.get("id")

    try:
        result = await self._handle(str(method), params)
    except ProtocolError as exc:
        resp = JsonRpcResponse(
            id=req_id, error=JsonRpcError(exc.code or -32601, exc.message)
        )
        return JsonRpcCodec.encode_response(resp)
    except Exception as exc:  # noqa: BLE001
        resp = JsonRpcResponse(id=req_id, error=JsonRpcError(-32603, str(exc)))
        return JsonRpcCodec.encode_response(resp)

    if req_id is None:
        return None  # notification, no response
    return JsonRpcCodec.encode_response(JsonRpcResponse(id=req_id, result=result))

make_request staticmethod

Python
make_request(method: str, params: dict[str, Any] | None = None, id_: str | int | None = None) -> str
Source code in apogee_ai_comunication/infrastructure/mcp/native_mcp_server.py
Python
@staticmethod
def make_request(
    method: str, params: dict[str, Any] | None = None, id_: str | int | None = None
) -> str:
    return JsonRpcCodec.encode_request(
        JsonRpcRequest(method=method, params=params or {}, id=id_)
    )

NativeUiProvider

Python
NativeUiProvider()

Bases: IMcpUiProvider

Source code in apogee_ai_comunication/infrastructure/mcp_ui/native_ui_provider.py
Python
def __init__(self) -> None:
    self._resources: dict[str, ResourceDescriptor] = {}

register_ui

Python
register_ui(resource: ResourceDescriptor) -> None
Source code in apogee_ai_comunication/infrastructure/mcp_ui/native_ui_provider.py
Python
def register_ui(self, resource: ResourceDescriptor) -> None:
    if not resource.uri.startswith("ui://"):
        raise ProtocolError(
            f"MCP-UI resources must use the 'ui://' scheme, got: {resource.uri}"
        )
    self._resources[resource.uri] = resource

serve async

Python
serve(uri: str) -> ResourceDescriptor
Source code in apogee_ai_comunication/infrastructure/mcp_ui/native_ui_provider.py
Python
async def serve(self, uri: str) -> ResourceDescriptor:
    res = self._resources.get(uri)
    if res is None:
        raise ProtocolError(f"unknown ui resource: {uri}")
    return res

NoOpObservabilityEmitter

Bases: IObservabilityEmitter

emit

Python
emit(name: str, payload: dict[str, object]) -> None
Source code in apogee_ai_comunication/infrastructure/observability/noop_observability_emitter.py
Python
def emit(self, name: str, payload: dict[str, object]) -> None:
    return None

PromptTemplate dataclass

Python
PromptTemplate(name: str, description: str = '', arguments: dict[str, object] = dict(), template: str = '')

name instance-attribute

Python
name: str

description class-attribute instance-attribute

Python
description: str = ''

arguments class-attribute instance-attribute

Python
arguments: dict[str, object] = field(default_factory=dict)

template class-attribute instance-attribute

Python
template: str = ''

ReactPayloadEmitter

Python
ReactPayloadEmitter(renderer: NativeA2uiRenderer | None = None)
Source code in apogee_ai_comunication/infrastructure/a2ui/react_payload_emitter.py
Python
def __init__(self, renderer: NativeA2uiRenderer | None = None) -> None:
    self._renderer = renderer or NativeA2uiRenderer()

emit

Python
emit(root: UiComponent) -> str
Source code in apogee_ai_comunication/infrastructure/a2ui/react_payload_emitter.py
Python
def emit(self, root: UiComponent) -> str:
    return json.dumps(self._renderer.render(root), ensure_ascii=False)

ResourceDescriptor dataclass

Python
ResourceDescriptor(uri: str, name: str, mime_type: str = 'text/plain', description: str = '', contents: str | bytes | None = None)

uri instance-attribute

Python
uri: str

name instance-attribute

Python
name: str

mime_type class-attribute instance-attribute

Python
mime_type: str = 'text/plain'

description class-attribute instance-attribute

Python
description: str = ''

contents class-attribute instance-attribute

Python
contents: str | bytes | None = None

ServerConfig dataclass

Python
ServerConfig(host: str = '127.0.0.1', port: int = 8000, transport: TransportKind = HTTP_SSE, auth_scheme: AuthScheme = NONE, auth_token: str | None = None, public_url: str | None = None, extras: dict[str, object] = dict())

host class-attribute instance-attribute

Python
host: str = '127.0.0.1'

port class-attribute instance-attribute

Python
port: int = 8000

transport class-attribute instance-attribute

Python
transport: TransportKind = HTTP_SSE

auth_scheme class-attribute instance-attribute

Python
auth_scheme: AuthScheme = NONE

auth_token class-attribute instance-attribute

Python
auth_token: str | None = None

public_url class-attribute instance-attribute

Python
public_url: str | None = None

extras class-attribute instance-attribute

Python
extras: dict[str, object] = field(default_factory=dict)

StreamingConfig dataclass

Python
StreamingConfig(chunk_size: int = 1024, flush_interval_seconds: float = 0.05, backpressure_limit: int = 256)

chunk_size class-attribute instance-attribute

Python
chunk_size: int = 1024

flush_interval_seconds class-attribute instance-attribute

Python
flush_interval_seconds: float = 0.05

backpressure_limit class-attribute instance-attribute

Python
backpressure_limit: int = 256

TaskArtifact dataclass

Python
TaskArtifact(name: str, mime_type: str = 'application/json', data: dict[str, object] = dict(), url: str | None = None)

A produced output (file, structured data, link).

name instance-attribute

Python
name: str

mime_type class-attribute instance-attribute

Python
mime_type: str = 'application/json'

data class-attribute instance-attribute

Python
data: dict[str, object] = field(default_factory=dict)

url class-attribute instance-attribute

Python
url: str | None = None

TaskState

Bases: str, Enum

SUBMITTED class-attribute instance-attribute

Python
SUBMITTED = 'submitted'

WORKING class-attribute instance-attribute

Python
WORKING = 'working'

INPUT_REQUIRED class-attribute instance-attribute

Python
INPUT_REQUIRED = 'input_required'

COMPLETED class-attribute instance-attribute

Python
COMPLETED = 'completed'

FAILED class-attribute instance-attribute

Python
FAILED = 'failed'

CANCELED class-attribute instance-attribute

Python
CANCELED = 'canceled'

ToolDefinition dataclass

Python
ToolDefinition(name: str, description: str = '', input_schema: dict[str, object] = dict(), output_schema: dict[str, object] | None = None, annotations: dict[str, object] = dict())

JSON-schema based tool definition shared across protocols.

name instance-attribute

Python
name: str

description class-attribute instance-attribute

Python
description: str = ''

input_schema class-attribute instance-attribute

Python
input_schema: dict[str, object] = field(default_factory=dict)

output_schema class-attribute instance-attribute

Python
output_schema: dict[str, object] | None = None

annotations class-attribute instance-attribute

Python
annotations: dict[str, object] = field(default_factory=dict)

ToolInvocation dataclass

Python
ToolInvocation(name: str, arguments: dict[str, object] = dict(), call_id: str | None = None)

name instance-attribute

Python
name: str

arguments class-attribute instance-attribute

Python
arguments: dict[str, object] = field(default_factory=dict)

call_id class-attribute instance-attribute

Python
call_id: str | None = None

ToolMapper

def_to_dto staticmethod

Python
def_to_dto(d: ToolDefinition) -> ToolDefinitionDTO
Source code in apogee_ai_comunication/application/mappers/tool_mapper.py
Python
@staticmethod
def def_to_dto(d: ToolDefinition) -> ToolDefinitionDTO:
    return ToolDefinitionDTO(
        name=d.name,
        description=d.description,
        input_schema=dict(d.input_schema),
        output_schema=dict(d.output_schema) if d.output_schema else None,
        annotations=dict(d.annotations),
    )

def_from_dto staticmethod

Python
def_from_dto(dto: ToolDefinitionDTO) -> ToolDefinition
Source code in apogee_ai_comunication/application/mappers/tool_mapper.py
Python
@staticmethod
def def_from_dto(dto: ToolDefinitionDTO) -> ToolDefinition:
    return ToolDefinition(
        name=dto.name,
        description=dto.description,
        input_schema=dict(dto.input_schema),
        output_schema=dict(dto.output_schema) if dto.output_schema else None,
        annotations=dict(dto.annotations),
    )

inv_to_dto staticmethod

Python
inv_to_dto(i: ToolInvocation) -> ToolInvocationDTO
Source code in apogee_ai_comunication/application/mappers/tool_mapper.py
Python
@staticmethod
def inv_to_dto(i: ToolInvocation) -> ToolInvocationDTO:
    return ToolInvocationDTO(
        name=i.name, arguments=dict(i.arguments), call_id=i.call_id
    )

inv_from_dto staticmethod

Python
inv_from_dto(dto: ToolInvocationDTO) -> ToolInvocation
Source code in apogee_ai_comunication/application/mappers/tool_mapper.py
Python
@staticmethod
def inv_from_dto(dto: ToolInvocationDTO) -> ToolInvocation:
    return ToolInvocation(
        name=dto.name, arguments=dict(dto.arguments), call_id=dto.call_id
    )

res_to_dto staticmethod

Python
res_to_dto(r: ToolResult) -> ToolResultDTO
Source code in apogee_ai_comunication/application/mappers/tool_mapper.py
Python
@staticmethod
def res_to_dto(r: ToolResult) -> ToolResultDTO:
    return ToolResultDTO(
        call_id=r.call_id,
        content=r.content,
        is_error=r.is_error,
        error_message=r.error_message,
    )

res_from_dto staticmethod

Python
res_from_dto(dto: ToolResultDTO) -> ToolResult
Source code in apogee_ai_comunication/application/mappers/tool_mapper.py
Python
@staticmethod
def res_from_dto(dto: ToolResultDTO) -> ToolResult:
    return ToolResult(
        call_id=dto.call_id,
        content=dto.content,
        is_error=dto.is_error,
        error_message=dto.error_message,
    )

ToolResult dataclass

Python
ToolResult(call_id: str | None, content: object, is_error: bool = False, error_message: str | None = None)

call_id instance-attribute

Python
call_id: str | None

content instance-attribute

Python
content: object

is_error class-attribute instance-attribute

Python
is_error: bool = False

error_message class-attribute instance-attribute

Python
error_message: str | None = None

TransportConfig dataclass

Python
TransportConfig(kind: TransportKind = HTTP_SSE, keepalive_seconds: float = 15.0, max_message_bytes: int = 4 * 1024 * 1024)

kind class-attribute instance-attribute

Python
kind: TransportKind = HTTP_SSE

keepalive_seconds class-attribute instance-attribute

Python
keepalive_seconds: float = 15.0

max_message_bytes class-attribute instance-attribute

Python
max_message_bytes: int = 4 * 1024 * 1024

UiComponent dataclass

Python
UiComponent(type: str, props: dict[str, object] = dict(), children: tuple['UiComponent', ...] = (), binding: str | None = None)

Node in a declarative UI tree (e.g. A2UI Card, Button, TextField).

type instance-attribute

Python
type: str

props class-attribute instance-attribute

Python
props: dict[str, object] = field(default_factory=dict)

children class-attribute instance-attribute

Python
children: tuple['UiComponent', ...] = ()

binding class-attribute instance-attribute

Python
binding: str | None = None

UiEvent dataclass

Python
UiEvent(kind: UiEventKind, payload: dict[str, object] = dict(), sequence: int | None = None)

A single event emitted on the AG-UI stream.

kind instance-attribute

Python
kind: UiEventKind

payload class-attribute instance-attribute

Python
payload: dict[str, object] = field(default_factory=dict)

sequence class-attribute instance-attribute

Python
sequence: int | None = None

UiEventMapper

to_dto staticmethod

Python
to_dto(e: UiEvent) -> UiEventDTO
Source code in apogee_ai_comunication/application/mappers/ui_event_mapper.py
Python
@staticmethod
def to_dto(e: UiEvent) -> UiEventDTO:
    return UiEventDTO(kind=e.kind.value, payload=dict(e.payload), sequence=e.sequence)

from_dto staticmethod

Python
from_dto(dto: UiEventDTO) -> UiEvent
Source code in apogee_ai_comunication/application/mappers/ui_event_mapper.py
Python
@staticmethod
def from_dto(dto: UiEventDTO) -> UiEvent:
    return UiEvent(
        kind=UiEventKind(dto.kind), payload=dict(dto.payload), sequence=dto.sequence
    )

finish_event

Python
finish_event(run_id: str, *, status: str = 'ok') -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
def finish_event(run_id: str, *, status: str = "ok") -> UiEvent:
    return UiEvent(
        kind=UiEventKind.LIFECYCLE,
        payload={"phase": "finish", "run_id": run_id, "status": status},
    )

start_event

Python
start_event(run_id: str) -> UiEvent
Source code in apogee_ai_comunication/infrastructure/ag_ui/lifecycle.py
Python
def start_event(run_id: str) -> UiEvent:
    return UiEvent(kind=UiEventKind.LIFECYCLE, payload={"phase": "start", "run_id": run_id})

Other · DTOs

AgentCardDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='allow')

name instance-attribute

Python
name: str

version class-attribute instance-attribute

Python
version: str = '0.1.0'

description class-attribute instance-attribute

Python
description: str = ''

url class-attribute instance-attribute

Python
url: str | None = None

protocols class-attribute instance-attribute

Python
protocols: list[str] = Field(default_factory=list)

capabilities class-attribute instance-attribute

Python
capabilities: list[AgentSkillDTO] = Field(default_factory=list)

auth_schemes class-attribute instance-attribute

Python
auth_schemes: list[str] = Field(default_factory=lambda: ['none'])

AgentMessageDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='allow')

role instance-attribute

Python
role: str

parts class-attribute instance-attribute

Python
parts: list[MessagePartDTO] = Field(default_factory=list)

name class-attribute instance-attribute

Python
name: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, object] = Field(default_factory=dict)

AgentSkillDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

name instance-attribute

Python
name: str

description class-attribute instance-attribute

Python
description: str = ''

input_modes class-attribute instance-attribute

Python
input_modes: list[str] = Field(default_factory=lambda: ['text'])

output_modes class-attribute instance-attribute

Python
output_modes: list[str] = Field(default_factory=lambda: ['text'])

tags class-attribute instance-attribute

Python
tags: list[str] = Field(default_factory=list)

AgentTaskDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='allow')

task_id instance-attribute

Python
task_id: str

state instance-attribute

Python
state: str

history class-attribute instance-attribute

Python
history: list[AgentMessageDTO] = Field(default_factory=list)

artifacts class-attribute instance-attribute

Python
artifacts: list[TaskArtifactDTO] = Field(default_factory=list)

metadata class-attribute instance-attribute

Python
metadata: dict[str, object] = Field(default_factory=dict)

MessagePartDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

kind instance-attribute

Python
kind: str

text class-attribute instance-attribute

Python
text: str | None = None

data class-attribute instance-attribute

Python
data: dict[str, object] | None = None

mime_type class-attribute instance-attribute

Python
mime_type: str | None = None

TaskArtifactDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

name instance-attribute

Python
name: str

mime_type class-attribute instance-attribute

Python
mime_type: str = 'application/json'

data class-attribute instance-attribute

Python
data: dict[str, object] = Field(default_factory=dict)

url class-attribute instance-attribute

Python
url: str | None = None

ToolDefinitionDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='allow')

name instance-attribute

Python
name: str

description class-attribute instance-attribute

Python
description: str = ''

input_schema class-attribute instance-attribute

Python
input_schema: dict[str, object] = Field(default_factory=dict)

output_schema class-attribute instance-attribute

Python
output_schema: dict[str, object] | None = None

annotations class-attribute instance-attribute

Python
annotations: dict[str, object] = Field(default_factory=dict)

ToolInvocationDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

name instance-attribute

Python
name: str

arguments class-attribute instance-attribute

Python
arguments: dict[str, object] = Field(default_factory=dict)

call_id class-attribute instance-attribute

Python
call_id: str | None = None

ToolResultDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='allow')

call_id class-attribute instance-attribute

Python
call_id: str | None = None

content class-attribute instance-attribute

Python
content: object = None

is_error class-attribute instance-attribute

Python
is_error: bool = False

error_message class-attribute instance-attribute

Python
error_message: str | None = None

UiEventDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='allow')

kind instance-attribute

Python
kind: str

payload class-attribute instance-attribute

Python
payload: dict[str, object] = Field(default_factory=dict)

sequence class-attribute instance-attribute

Python
sequence: int | None = None

Other · Enums

ProtocolKind

Bases: str, Enum

Identifies which protocol a server/client speaks.

MCP class-attribute instance-attribute

Python
MCP = 'mcp'

A2A class-attribute instance-attribute

Python
A2A = 'a2a'

AG_UI class-attribute instance-attribute

Python
AG_UI = 'ag_ui'

ACP class-attribute instance-attribute

Python
ACP = 'acp'

A2UI class-attribute instance-attribute

Python
A2UI = 'a2ui'

MCP_UI class-attribute instance-attribute

Python
MCP_UI = 'mcp_ui'

TransportKind

Bases: str, Enum

STDIO class-attribute instance-attribute

Python
STDIO = 'stdio'

HTTP_SSE class-attribute instance-attribute

Python
HTTP_SSE = 'http_sse'

HTTP_STREAM class-attribute instance-attribute

Python
HTTP_STREAM = 'http_stream'

WEBSOCKET class-attribute instance-attribute

Python
WEBSOCKET = 'websocket'

GRPC class-attribute instance-attribute

Python
GRPC = 'grpc'

UiEventKind

Bases: str, Enum

TEXT_DELTA class-attribute instance-attribute

Python
TEXT_DELTA = 'text_delta'

TOOL_CALL class-attribute instance-attribute

Python
TOOL_CALL = 'tool_call'

TOOL_RESULT class-attribute instance-attribute

Python
TOOL_RESULT = 'tool_result'

STATE_PATCH class-attribute instance-attribute

Python
STATE_PATCH = 'state_patch'

LIFECYCLE class-attribute instance-attribute

Python
LIFECYCLE = 'lifecycle'

ERROR class-attribute instance-attribute

Python
ERROR = 'error'

Other · Exceptions

AiCommException

Python
AiCommException(message: str, *, cause: BaseException | None = None)

Bases: Exception

Root of every exception raised by apogee_ai_comunication.

Source code in apogee_ai_comunication/domain/exceptions/ai_comm_exception.py
Python
def __init__(self, message: str, *, cause: BaseException | None = None) -> None:
    super().__init__(message)
    self.message = message
    self.cause = cause

message instance-attribute

Python
message = message

cause instance-attribute

Python
cause = cause

AuthError

Python
AuthError(message: str, *, cause: BaseException | None = None)

Bases: AiCommException

Raised when credentials are missing, expired or insufficient.

Source code in apogee_ai_comunication/domain/exceptions/ai_comm_exception.py
Python
def __init__(self, message: str, *, cause: BaseException | None = None) -> None:
    super().__init__(message)
    self.message = message
    self.cause = cause

ProtocolError

Python
ProtocolError(message: str, *, code: int | None = None, cause: BaseException | None = None)

Bases: AiCommException

Raised when a payload violates the protocol contract.

Source code in apogee_ai_comunication/domain/exceptions/protocol_error.py
Python
def __init__(
    self,
    message: str,
    *,
    code: int | None = None,
    cause: BaseException | None = None,
) -> None:
    super().__init__(message, cause=cause)
    self.code = code

code instance-attribute

Python
code = code

ProviderNotInstalledError

Python
ProviderNotInstalledError(provider: str, extras_name: str)

Bases: AiCommException

The optional dependency that backs an adapter is not installed.

Source code in apogee_ai_comunication/domain/exceptions/provider_not_installed_error.py
Python
def __init__(self, provider: str, extras_name: str) -> None:
    super().__init__(
        f"Provider '{provider}' is not installed. "
        f"Install with: pip install 'apogee-ai-comunication[{extras_name}]'"
    )
    self.provider = provider
    self.extras_name = extras_name

provider instance-attribute

Python
provider = provider

extras_name instance-attribute

Python
extras_name = extras_name

ToolInvocationError

Python
ToolInvocationError(message: str, *, tool_name: str, cause: BaseException | None = None)

Bases: AiCommException

The tool returned an error or could not be invoked.

Source code in apogee_ai_comunication/domain/exceptions/tool_invocation_error.py
Python
def __init__(
    self,
    message: str,
    *,
    tool_name: str,
    cause: BaseException | None = None,
) -> None:
    super().__init__(message, cause=cause)
    self.tool_name = tool_name

tool_name instance-attribute

Python
tool_name = tool_name

TransportError

Python
TransportError(message: str, *, cause: BaseException | None = None)

Bases: AiCommException

Network, encoding or serialization failure on the transport layer.

Source code in apogee_ai_comunication/domain/exceptions/ai_comm_exception.py
Python
def __init__(self, message: str, *, cause: BaseException | None = None) -> None:
    super().__init__(message)
    self.message = message
    self.cause = cause

Other · Protocols (ports)

IA2aClient

Bases: Protocol

discover async

Python
discover() -> AgentCard
Source code in apogee_ai_comunication/domain/services/i_a2a_client.py
Python
async def discover(self) -> AgentCard: ...

send_task async

Python
send_task(message: AgentMessage) -> AgentTask
Source code in apogee_ai_comunication/domain/services/i_a2a_client.py
Python
async def send_task(self, message: AgentMessage) -> AgentTask: ...

watch_task async

Python
watch_task(task_id: str) -> AsyncIterator[AgentTask]
Source code in apogee_ai_comunication/domain/services/i_a2a_client.py
Python
async def watch_task(self, task_id: str) -> AsyncIterator[AgentTask]: ...

close async

Python
close() -> None
Source code in apogee_ai_comunication/domain/services/i_a2a_client.py
Python
async def close(self) -> None: ...

IA2aServer

Bases: Protocol

agent_card property

Python
agent_card: AgentCard

handle_task async

Python
handle_task(message: AgentMessage) -> AgentTask
Source code in apogee_ai_comunication/domain/services/i_a2a_server.py
Python
async def handle_task(self, message: AgentMessage) -> AgentTask: ...

stream_task async

Python
stream_task(task_id: str) -> AsyncIterator[AgentTask]
Source code in apogee_ai_comunication/domain/services/i_a2a_server.py
Python
async def stream_task(self, task_id: str) -> AsyncIterator[AgentTask]: ...

run async

Python
run() -> None
Source code in apogee_ai_comunication/domain/services/i_a2a_server.py
Python
async def run(self) -> None: ...

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/domain/services/i_a2a_server.py
Python
async def stop(self) -> None: ...

IA2uiRenderer

Bases: Protocol

render

Python
render(root: UiComponent) -> dict[str, object]
Source code in apogee_ai_comunication/domain/services/i_a2ui_renderer.py
Python
def render(self, root: UiComponent) -> dict[str, object]: ...

validate

Python
validate(root: UiComponent) -> list[str]
Source code in apogee_ai_comunication/domain/services/i_a2ui_renderer.py
Python
def validate(self, root: UiComponent) -> list[str]: ...

IAcpClient

Bases: Protocol

send async

Python
send(message: AgentMessage) -> AgentMessage
Source code in apogee_ai_comunication/domain/services/i_acp_client.py
Python
async def send(self, message: AgentMessage) -> AgentMessage: ...

close async

Python
close() -> None
Source code in apogee_ai_comunication/domain/services/i_acp_client.py
Python
async def close(self) -> None: ...

IAcpServer

Bases: Protocol

receive async

Python
receive(message: AgentMessage) -> AgentMessage
Source code in apogee_ai_comunication/domain/services/i_acp_server.py
Python
async def receive(self, message: AgentMessage) -> AgentMessage: ...

run async

Python
run() -> None
Source code in apogee_ai_comunication/domain/services/i_acp_server.py
Python
async def run(self) -> None: ...

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/domain/services/i_acp_server.py
Python
async def stop(self) -> None: ...

IAgUiEventEmitter

Bases: Protocol

emit async

Python
emit(event: UiEvent) -> None
Source code in apogee_ai_comunication/domain/services/i_ag_ui_event_emitter.py
Python
async def emit(self, event: UiEvent) -> None: ...

close async

Python
close() -> None
Source code in apogee_ai_comunication/domain/services/i_ag_ui_event_emitter.py
Python
async def close(self) -> None: ...

IAgUiServer

Bases: Protocol

stream_events async

Python
stream_events(message: AgentMessage) -> AsyncIterator[UiEvent]
Source code in apogee_ai_comunication/domain/services/i_ag_ui_server.py
Python
async def stream_events(self, message: AgentMessage) -> AsyncIterator[UiEvent]: ...

run async

Python
run() -> None
Source code in apogee_ai_comunication/domain/services/i_ag_ui_server.py
Python
async def run(self) -> None: ...

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/domain/services/i_ag_ui_server.py
Python
async def stop(self) -> None: ...

IMcpClient

Bases: Protocol

list_tools async

Python
list_tools() -> list[ToolDefinition]
Source code in apogee_ai_comunication/domain/services/i_mcp_client.py
Python
async def list_tools(self) -> list[ToolDefinition]: ...

call_tool async

Python
call_tool(invocation: ToolInvocation) -> ToolResult
Source code in apogee_ai_comunication/domain/services/i_mcp_client.py
Python
async def call_tool(self, invocation: ToolInvocation) -> ToolResult: ...

get_resource async

Python
get_resource(uri: str) -> ResourceDescriptor
Source code in apogee_ai_comunication/domain/services/i_mcp_client.py
Python
async def get_resource(self, uri: str) -> ResourceDescriptor: ...

close async

Python
close() -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_client.py
Python
async def close(self) -> None: ...

IMcpServer

Bases: Protocol

register_tool

Python
register_tool(definition: ToolDefinition, handler: ToolHandler) -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_server.py
Python
def register_tool(self, definition: ToolDefinition, handler: ToolHandler) -> None: ...

register_resource

Python
register_resource(resource: ResourceDescriptor) -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_server.py
Python
def register_resource(self, resource: ResourceDescriptor) -> None: ...

register_prompt

Python
register_prompt(prompt: PromptTemplate) -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_server.py
Python
def register_prompt(self, prompt: PromptTemplate) -> None: ...

run async

Python
run() -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_server.py
Python
async def run(self) -> None: ...

stop async

Python
stop() -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_server.py
Python
async def stop(self) -> None: ...

IMcpUiProvider

Bases: Protocol

register_ui

Python
register_ui(resource: ResourceDescriptor) -> None
Source code in apogee_ai_comunication/domain/services/i_mcp_ui_provider.py
Python
def register_ui(self, resource: ResourceDescriptor) -> None: ...

serve async

Python
serve(uri: str) -> ResourceDescriptor
Source code in apogee_ai_comunication/domain/services/i_mcp_ui_provider.py
Python
async def serve(self, uri: str) -> ResourceDescriptor: ...

IMessageRouter

Bases: Protocol

route async

Python
route(target: ProtocolKind, message: AgentMessage) -> AgentMessage
Source code in apogee_ai_comunication/domain/services/i_message_router.py
Python
async def route(self, target: ProtocolKind, message: AgentMessage) -> AgentMessage: ...

IObservabilityEmitter

Bases: Protocol

emit

Python
emit(name: str, payload: dict[str, object]) -> None
Source code in apogee_ai_comunication/domain/services/i_observability_emitter.py
Python
def emit(self, name: str, payload: dict[str, object]) -> None: ...

IToolRegistry

Bases: Protocol

register

Python
register(definition: ToolDefinition, handler: ToolHandler) -> None
Source code in apogee_ai_comunication/domain/services/i_tool_registry.py
Python
def register(self, definition: ToolDefinition, handler: ToolHandler) -> None: ...

list_tools

Python
list_tools() -> list[ToolDefinition]
Source code in apogee_ai_comunication/domain/services/i_tool_registry.py
Python
def list_tools(self) -> list[ToolDefinition]: ...

get_handler

Python
get_handler(name: str) -> ToolHandler | None
Source code in apogee_ai_comunication/domain/services/i_tool_registry.py
Python
def get_handler(self, name: str) -> ToolHandler | None: ...

invoke async

Python
invoke(invocation: ToolInvocation) -> ToolResult
Source code in apogee_ai_comunication/domain/services/i_tool_registry.py
Python
async def invoke(self, invocation: ToolInvocation) -> ToolResult: ...

Other · Use cases

InvokeMcpToolUseCase dataclass

Python
InvokeMcpToolUseCase(client: IMcpClient)

client instance-attribute

Python
client: IMcpClient

execute async

Python
execute(invocation: ToolInvocation) -> ToolResult
Source code in apogee_ai_comunication/application/use_cases/invoke_mcp_tool_use_case.py
Python
async def execute(self, invocation: ToolInvocation) -> ToolResult:
    return await self.client.call_tool(invocation)

PublishAgentCardUseCase dataclass

Python
PublishAgentCardUseCase(server: IA2aServer)

server instance-attribute

Python
server: IA2aServer

execute

Python
execute() -> AgentCardDTO
Source code in apogee_ai_comunication/application/use_cases/publish_agent_card_use_case.py
Python
def execute(self) -> AgentCardDTO:
    return AgentCardMapper.to_dto(self.server.agent_card)

RenderA2uiTreeUseCase dataclass

Python
RenderA2uiTreeUseCase(renderer: IA2uiRenderer)

renderer instance-attribute

Python
renderer: IA2uiRenderer

execute

Python
execute(root: UiComponent) -> dict[str, object]
Source code in apogee_ai_comunication/application/use_cases/render_a2ui_tree_use_case.py
Python
def execute(self, root: UiComponent) -> dict[str, object]:
    problems = self.renderer.validate(root)
    if problems:
        raise ProtocolError(
            "A2UI tree validation failed: " + "; ".join(problems)
        )
    return self.renderer.render(root)

RouteMessageUseCase dataclass

Python
RouteMessageUseCase(router: IMessageRouter)

router instance-attribute

Python
router: IMessageRouter

execute async

Python
execute(target: ProtocolKind, message: AgentMessage) -> AgentMessage
Source code in apogee_ai_comunication/application/use_cases/route_message_use_case.py
Python
async def execute(
    self, target: ProtocolKind, message: AgentMessage
) -> AgentMessage:
    return await self.router.route(target, message)

SendA2aTaskUseCase dataclass

Python
SendA2aTaskUseCase(client: IA2aClient)

client instance-attribute

Python
client: IA2aClient

execute async

Python
execute(message: AgentMessage) -> AsyncIterator[AgentTask]
Source code in apogee_ai_comunication/application/use_cases/send_a2a_task_use_case.py
Python
async def execute(self, message: AgentMessage) -> AsyncIterator[AgentTask]:
    task = await self.client.send_task(message)
    async for update in self.client.watch_task(task.task_id):
        yield update

ServeMcpUiResourceUseCase dataclass

Python
ServeMcpUiResourceUseCase(provider: IMcpUiProvider)

provider instance-attribute

Python
provider: IMcpUiProvider

execute async

Python
execute(uri: str) -> ResourceDescriptor
Source code in apogee_ai_comunication/application/use_cases/serve_mcp_ui_resource_use_case.py
Python
async def execute(self, uri: str) -> ResourceDescriptor:
    return await self.provider.serve(uri)

StartMcpServerUseCase dataclass

Python
StartMcpServerUseCase(server: IMcpServer)

server instance-attribute

Python
server: IMcpServer

execute async

Python
execute() -> None
Source code in apogee_ai_comunication/application/use_cases/start_mcp_server_use_case.py
Python
async def execute(self) -> None:
    await self.server.run()

StreamAgUiEventsUseCase dataclass

Python
StreamAgUiEventsUseCase(server: IAgUiServer)

server instance-attribute

Python
server: IAgUiServer

execute async

Python
execute(message: AgentMessage) -> AsyncIterator[UiEvent]
Source code in apogee_ai_comunication/application/use_cases/stream_ag_ui_events_use_case.py
Python
async def execute(self, message: AgentMessage) -> AsyncIterator[UiEvent]:
    async for event in self.server.stream_events(message):
        yield event