Skip to content

API reference

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

Application · DTOs

BenchDTO dataclass

Python
BenchDTO(messages: int = 200)

messages class-attribute instance-attribute

Python
messages: int = 200

BroadcastDTO dataclass

Python
BroadcastDTO(body: str, connectors: tuple[str, ...] = ())

body instance-attribute

Python
body: str

connectors class-attribute instance-attribute

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

SendDTO dataclass

Python
SendDTO(body: str, channel: str = '', subject: str = '', connector: str = 'console')

body instance-attribute

Python
body: str

channel class-attribute instance-attribute

Python
channel: str = ''

subject class-attribute instance-attribute

Python
subject: str = ''

connector class-attribute instance-attribute

Python
connector: str = 'console'

WebhookDTO dataclass

Python
WebhookDTO(url: str, body: str)

url instance-attribute

Python
url: str

body instance-attribute

Python
body: str

Application · Use cases

BenchSendUseCase

execute async

Python
execute(messages: int) -> dict[str, float]
Source code in apogee_ai_connectors/application/use_cases/bench_send_use_case.py
Python
async def execute(self, messages: int) -> dict[str, float]:
    if messages <= 0:
        raise ValueError("messages must be positive")
    connector = ConsoleConnector(stream=io.StringIO())
    start = time.perf_counter()
    for i in range(messages):
        await connector.send(ConnectorMessage(
            channel="#bench", body=f"msg-{i}",
        ))
    elapsed = (time.perf_counter() - start) * 1000.0
    return {
        "messages": float(messages),
        "elapsed_ms": elapsed,
        "messages_per_second": (messages / elapsed * 1000.0) if elapsed > 0 else 0.0,
    }

BroadcastUseCase

Python
BroadcastUseCase(registry)

Sends a message through every connector in parallel.

Source code in apogee_ai_connectors/application/use_cases/broadcast_use_case.py
Python
def __init__(self, registry) -> None:
    self._registry = registry

execute async

Python
execute(message: ConnectorMessage, names: tuple[str, ...] = ()) -> list[ConnectorEvent]
Source code in apogee_ai_connectors/application/use_cases/broadcast_use_case.py
Python
async def execute(
    self, message: ConnectorMessage, names: tuple[str, ...] = (),
) -> list[ConnectorEvent]:
    targets = (
        [self._registry.get(n) for n in names]
        if names else list(self._registry.all())
    )
    if not targets:
        return []
    return list(await asyncio.gather(
        *[c.send(message) for c in targets],
        return_exceptions=False,
    ))

ListConnectorsUseCase

Python
ListConnectorsUseCase(registry)
Source code in apogee_ai_connectors/application/use_cases/list_connectors_use_case.py
Python
def __init__(self, registry) -> None:
    self._registry = registry

execute async

Python
execute() -> list[str]
Source code in apogee_ai_connectors/application/use_cases/list_connectors_use_case.py
Python
async def execute(self) -> list[str]:
    return self._registry.list()

SendUseCase

Python
SendUseCase(connector)
Source code in apogee_ai_connectors/application/use_cases/send_use_case.py
Python
def __init__(self, connector) -> None:
    self._connector = connector

execute async

Python
execute(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/application/use_cases/send_use_case.py
Python
async def execute(self, message: ConnectorMessage) -> ConnectorEvent:
    return await self._connector.send(message)

Domain

ConnectorEvent dataclass

Python
ConnectorEvent(connector: str, message_id: str, status: DeliveryStatus, latency_ms: float = 0.0, error: str = '', timestamp_s: float = time(), metadata: dict[str, str] = dict())

connector instance-attribute

Python
connector: str

message_id instance-attribute

Python
message_id: str

status instance-attribute

Python
status: DeliveryStatus

latency_ms class-attribute instance-attribute

Python
latency_ms: float = 0.0

error class-attribute instance-attribute

Python
error: str = ''

timestamp_s class-attribute instance-attribute

Python
timestamp_s: float = field(default_factory=time)

metadata class-attribute instance-attribute

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

ConnectorMessage dataclass

Python
ConnectorMessage(body: str, channel: str = '', subject: str = '', recipients: tuple[str, ...] = (), headers: dict[str, str] = dict(), metadata: dict[str, str] = dict(), id: str = (lambda: f'm-{hex[:8]}')())

body instance-attribute

Python
body: str

channel class-attribute instance-attribute

Python
channel: str = ''

subject class-attribute instance-attribute

Python
subject: str = ''

recipients class-attribute instance-attribute

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

headers class-attribute instance-attribute

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

metadata class-attribute instance-attribute

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

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: f'm-{hex[:8]}')

DeliveryStatus

Bases: str, Enum

DELIVERED class-attribute instance-attribute

Python
DELIVERED = 'delivered'

FAILED class-attribute instance-attribute

Python
FAILED = 'failed'

SKIPPED class-attribute instance-attribute

Python
SKIPPED = 'skipped'

Domain · Enums

ConnectorKind

Bases: str, Enum

CONSOLE class-attribute instance-attribute

Python
CONSOLE = 'console'

WEBHOOK class-attribute instance-attribute

Python
WEBHOOK = 'webhook'

EMAIL class-attribute instance-attribute

Python
EMAIL = 'email'

SLACK class-attribute instance-attribute

Python
SLACK = 'slack'

DISCORD class-attribute instance-attribute

Python
DISCORD = 'discord'

HTTP class-attribute instance-attribute

Python
HTTP = 'http'

Domain · Exceptions

ConnectorError

Bases: Exception

Base for apogee-ai-connectors errors.

ConnectorNotFoundException

Python
ConnectorNotFoundException(name: str)

Bases: ConnectorError

Source code in apogee_ai_connectors/domain/exceptions/connector_exceptions.py
Python
def __init__(self, name: str) -> None:
    super().__init__(f"Connector not found: {name!r}")
    self.name = name

name instance-attribute

Python
name = name

DeliveryError

Python
DeliveryError(connector: str, message: str)

Bases: ConnectorError

Source code in apogee_ai_connectors/domain/exceptions/connector_exceptions.py
Python
def __init__(self, connector: str, message: str) -> None:
    super().__init__(f"Connector {connector!r} delivery failed: {message}")
    self.connector = connector

connector instance-attribute

Python
connector = connector

Domain · Protocols (ports)

IConnector

Bases: Protocol

name instance-attribute

Python
name: str

send async

Python
send(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/domain/services/i_connector.py
Python
async def send(self, message: ConnectorMessage) -> ConnectorEvent: ...

IConnectorRegistry

Bases: Protocol

register

Python
register(connector) -> None
Source code in apogee_ai_connectors/domain/services/i_connector_registry.py
Python
def register(self, connector) -> None: ...

get

Python
get(name: str)
Source code in apogee_ai_connectors/domain/services/i_connector_registry.py
Python
def get(self, name: str): ...

list

Python
list() -> list[str]
Source code in apogee_ai_connectors/domain/services/i_connector_registry.py
Python
def list(self) -> list[str]: ...

IEventSink

Bases: Protocol

emit async

Python
emit(event: ConnectorEvent) -> None
Source code in apogee_ai_connectors/domain/services/i_event_sink.py
Python
async def emit(self, event: ConnectorEvent) -> None: ...

Infrastructure

ConnectorRegistry

Python
ConnectorRegistry(connectors: Iterable | None = None)
Source code in apogee_ai_connectors/infrastructure/registries/connector_registry.py
Python
def __init__(self, connectors: Iterable | None = None) -> None:
    self._connectors: dict[str, object] = {}
    for c in connectors or []:
        self._connectors[c.name] = c

register

Python
register(connector) -> None
Source code in apogee_ai_connectors/infrastructure/registries/connector_registry.py
Python
def register(self, connector) -> None:
    self._connectors[connector.name] = connector

get

Python
get(name: str)
Source code in apogee_ai_connectors/infrastructure/registries/connector_registry.py
Python
def get(self, name: str):
    if name not in self._connectors:
        raise ConnectorNotFoundException(name)
    return self._connectors[name]

list

Python
list() -> list[str]
Source code in apogee_ai_connectors/infrastructure/registries/connector_registry.py
Python
def list(self) -> list[str]:
    return list(self._connectors.keys())

all

Python
all() -> list
Source code in apogee_ai_connectors/infrastructure/registries/connector_registry.py
Python
def all(self) -> list:
    return list(self._connectors.values())

ConsoleConnector

Python
ConsoleConnector(stream: TextIO | None = None)

Prints messages to a stream. CI-safe — keeps a sent log.

Source code in apogee_ai_connectors/infrastructure/connectors/console_connector.py
Python
def __init__(self, stream: TextIO | None = None) -> None:
    self._stream = stream or sys.stdout
    self.sent: list[ConnectorMessage] = []

name class-attribute instance-attribute

Python
name = 'console'

sent instance-attribute

Python
sent: list[ConnectorMessage] = []

send async

Python
send(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/infrastructure/connectors/console_connector.py
Python
async def send(self, message: ConnectorMessage) -> ConnectorEvent:
    start = time.perf_counter()
    target = message.channel or "<no-channel>"
    prefix = f"[{self.name}] {target}"
    if message.subject:
        prefix = f"{prefix} | {message.subject}"
    # Fall back to current sys.stdout if the cached stream got closed
    # (e.g. pytest's capsys replaces sys.stdout between tests).
    stream = self._stream
    if getattr(stream, "closed", False):
        stream = sys.stdout
    print(f"{prefix}: {message.body}", file=stream)
    self.sent.append(message)
    return ConnectorEvent(
        connector=self.name,
        message_id=message.id,
        status=DeliveryStatus.DELIVERED,
        latency_ms=(time.perf_counter() - start) * 1000.0,
    )

DiscordWebhookConnector

Python
DiscordWebhookConnector(webhook_url: str, timeout_s: float = 10.0)

Posts to a Discord incoming webhook (no SDK required).

Source code in apogee_ai_connectors/infrastructure/connectors/discord_connector.py
Python
def __init__(self, webhook_url: str, timeout_s: float = 10.0) -> None:
    if not webhook_url:
        raise ValueError("webhook_url cannot be empty")
    self._url = webhook_url
    self._timeout = timeout_s

name class-attribute instance-attribute

Python
name = 'discord'

send async

Python
send(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/infrastructure/connectors/discord_connector.py
Python
async def send(self, message: ConnectorMessage) -> ConnectorEvent:
    start = time.perf_counter()
    payload = {"content": message.body}
    if message.metadata.get("username"):
        payload["username"] = message.metadata["username"]
    try:
        status = await asyncio.to_thread(self._post_sync, payload)
    except DeliveryError as exc:
        return ConnectorEvent(
            connector=self.name, message_id=message.id,
            status=DeliveryStatus.FAILED, error=str(exc),
            latency_ms=(time.perf_counter() - start) * 1000.0,
        )
    delivered = status in (200, 204)
    return ConnectorEvent(
        connector=self.name, message_id=message.id,
        status=DeliveryStatus.DELIVERED if delivered else DeliveryStatus.FAILED,
        latency_ms=(time.perf_counter() - start) * 1000.0,
        metadata={"http_status": str(status)},
        error="" if delivered else f"http {status}",
    )

SlackConnector

Python
SlackConnector(token: str, default_channel: str = '')

Lazy slack-sdk async adapter — install via extras=slack.

Source code in apogee_ai_connectors/infrastructure/connectors/slack_connector.py
Python
def __init__(self, token: str, default_channel: str = "") -> None:
    if not token:
        raise ValueError("slack token cannot be empty")
    self._token = token
    self._default_channel = default_channel
    self._client = None

name class-attribute instance-attribute

Python
name = 'slack'

send async

Python
send(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/infrastructure/connectors/slack_connector.py
Python
async def send(self, message: ConnectorMessage) -> ConnectorEvent:
    self._ensure_client()
    start = time.perf_counter()
    channel = message.channel or self._default_channel
    if not channel:
        return ConnectorEvent(
            connector=self.name, message_id=message.id,
            status=DeliveryStatus.FAILED,
            error="no channel and no default_channel",
            latency_ms=(time.perf_counter() - start) * 1000.0,
        )
    try:
        await self._client.chat_postMessage(  # type: ignore[union-attr]
            channel=channel, text=message.body,
        )
    except Exception as exc:  # pragma: no cover
        return ConnectorEvent(
            connector=self.name, message_id=message.id,
            status=DeliveryStatus.FAILED, error=str(exc),
            latency_ms=(time.perf_counter() - start) * 1000.0,
        )
    return ConnectorEvent(
        connector=self.name, message_id=message.id,
        status=DeliveryStatus.DELIVERED,
        latency_ms=(time.perf_counter() - start) * 1000.0,
    )

SmtpEmailConnector

Python
SmtpEmailConnector(host: str, port: int = 587, username: str = '', password: str = '', sender: str = '', timeout_s: float = 10.0)

Sends an email via stdlib smtplib. Uses SMTP_SSL when port=465.

Source code in apogee_ai_connectors/infrastructure/connectors/smtp_email_connector.py
Python
def __init__(
    self,
    host: str,
    port: int = 587,
    username: str = "",
    password: str = "",
    sender: str = "",
    timeout_s: float = 10.0,
) -> None:
    if not host:
        raise ValueError("host cannot be empty")
    self._host = host
    self._port = port
    self._username = username
    self._password = password
    self._sender = sender or username
    self._timeout = timeout_s

name class-attribute instance-attribute

Python
name = 'email'

send async

Python
send(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/infrastructure/connectors/smtp_email_connector.py
Python
async def send(self, message: ConnectorMessage) -> ConnectorEvent:
    start = time.perf_counter()
    try:
        await asyncio.to_thread(self._send_sync, message)
    except DeliveryError as exc:
        return ConnectorEvent(
            connector=self.name, message_id=message.id,
            status=DeliveryStatus.FAILED, error=str(exc),
            latency_ms=(time.perf_counter() - start) * 1000.0,
        )
    return ConnectorEvent(
        connector=self.name, message_id=message.id,
        status=DeliveryStatus.DELIVERED,
        latency_ms=(time.perf_counter() - start) * 1000.0,
    )

WebhookConnector

Python
WebhookConnector(url: str, timeout_s: float = 10.0, headers: dict[str, str] | None = None)

POSTs message body as JSON to a fixed URL via urllib (no httpx required).

Source code in apogee_ai_connectors/infrastructure/connectors/webhook_connector.py
Python
def __init__(self, url: str, timeout_s: float = 10.0,
             headers: dict[str, str] | None = None) -> None:
    if not url:
        raise ValueError("url cannot be empty")
    self._url = url
    self._timeout = timeout_s
    self._headers = {
        "Content-Type": "application/json",
        **(headers or {}),
    }

name class-attribute instance-attribute

Python
name = 'webhook'

send async

Python
send(message: ConnectorMessage) -> ConnectorEvent
Source code in apogee_ai_connectors/infrastructure/connectors/webhook_connector.py
Python
async def send(self, message: ConnectorMessage) -> ConnectorEvent:
    start = time.perf_counter()
    try:
        data = json.loads(message.body)
        if not isinstance(data, dict):
            payload = {"body": message.body}
        else:
            payload = data
    except json.JSONDecodeError:
        payload = {"body": message.body}
    if message.channel:
        payload.setdefault("channel", message.channel)
    if message.subject:
        payload.setdefault("subject", message.subject)
    try:
        status = await asyncio.to_thread(self._post_sync, payload)
    except DeliveryError as exc:
        return ConnectorEvent(
            connector=self.name, message_id=message.id,
            status=DeliveryStatus.FAILED, error=str(exc),
            latency_ms=(time.perf_counter() - start) * 1000.0,
        )
    delivered = 200 <= status < 300
    return ConnectorEvent(
        connector=self.name, message_id=message.id,
        status=DeliveryStatus.DELIVERED if delivered else DeliveryStatus.FAILED,
        latency_ms=(time.perf_counter() - start) * 1000.0,
        metadata={"http_status": str(status)},
        error="" if delivered else f"http {status}",
    )