astropods-adapter-core

Framework-agnostic bridge between Python agents and the Astropods messaging service.

Installation

pip install astropods-adapter-core

Requires Python 3.10+.

Usage

If you're using a supported framework, use the pre-built adapter package instead (e.g. astropods-adapter-langchain). Use this package directly to connect a custom or unsupported framework.

Create a class with name, stream, and get_config, then call serve():

from astropods_adapter_core import StreamHooks, StreamOptions, serve

class MyAdapter:
    name = "My Agent"

    async def stream(self, prompt: str, hooks: StreamHooks, options: StreamOptions) -> None:
        try:
            hooks.on_chunk("Hello!")
            hooks.on_finish()
        except Exception as e:
            hooks.on_error(e)

    def get_config(self) -> dict:
        return {"system_prompt": "You are a helpful assistant.", "tools": []}

serve(MyAdapter())

serve() blocks until SIGINT or SIGTERM. Under ast dev, GRPC_SERVER_ADDR is injected automatically.

API

AgentAdapter protocol

StreamHooks

Call these inside stream() as the agent produces output:

on_trace_context(trace_context) is optional; call it with getattr when a turn has W3C trace context.

For CUSTOM status, include "custom_message" in the dict:

hooks.on_status_update({"status": "CUSTOM", "custom_message": "Fetching data..."})

create_traceparent(*, trace_id, span_id, trace_flags="01")

Formats a native trace/span ID pair into a W3C traceparent string to hand to on_trace_context. Returns "" for invalid or all-zero IDs, so it's safe to pass raw OpenTelemetry span context. trace_flags accepts an int or hex string and defaults to "01" (sampled).

from astropods_messaging import TraceContext
from astropods_adapter_core import create_traceparent

ctx = span.get_span_context()
traceparent = create_traceparent(
    trace_id=f"{ctx.trace_id:032x}",
    span_id=f"{ctx.span_id:016x}",
    trace_flags=int(ctx.trace_flags),
)
if traceparent:
    on_trace_context = getattr(hooks, "on_trace_context", None)
    if on_trace_context is not None:
        on_trace_context(TraceContext(traceparent=traceparent))

StreamOptions

Per-request context passed to stream():

Using platform_context

PlatformContext exposes the source-event fields adapters often need to branch on — the channel and thread to reply into, the workspace, the bot's own user ID, and an event_kind enum that distinguishes a DM from an @-mention from a thread reply without having to inspect message content.

from astropods_adapter_core import PlatformContext, StreamHooks, StreamOptions

async def stream(self, prompt, hooks: StreamHooks, options: StreamOptions) -> None:
    pc = options.platform_context
    if pc and pc.event_kind == PlatformContext.EVENT_KIND_APP_MENTION:
        hooks.on_chunk(f"You @-mentioned me in {pc.channel_name or pc.channel_id}.")
    else:
        hooks.on_chunk("Hello!")
    hooks.on_finish()

Always null-check first — platform_context is None for messages from the playground or direct gRPC clients. See PlatformContext for the full field list.

serve(adapter, options?)

Connects the adapter to the messaging service and blocks until shutdown.

from astropods_adapter_core import ServeOptions, serve

# Override the gRPC address (default: GRPC_SERVER_ADDR env var or localhost:9090)
serve(adapter, ServeOptions(server_address="astro-messaging:9090"))

MessagingBridge

serve() is a thin wrapper around MessagingBridge. Use it directly if you need lifecycle control:

import asyncio
from astropods_adapter_core import MessagingBridge

bridge = MessagingBridge(adapter)
asyncio.run(bridge.start())