Custom adapter (Python)

astropods-adapter-core is the framework-agnostic Python core. Implement the AgentAdapter protocol, hand it to serve(), and the bundled MessagingBridge handles the gRPC streaming loop, audio, feedback, and shutdown.

Use this when there’s no first-party adapter for your framework (or no framework at all — you’re calling LLM APIs directly).

Before you start

You need:

• An Astropods agent project with agent.interfaces.messaging: true in astropods.yml, so the messaging sidecar runs alongside your agent.
• Python 3.10 or newer.

Install

pip install astropods-adapter-core

Requires Python 3.10+.

Minimal adapter

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_chunk(f"{prompt}!")
            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(adapter) connects to localhost:9090 (or GRPC_SERVER_ADDR) and blocks until SIGINT / SIGTERM.

Verify it works

Enable the web adapter in astropods.yml (dev.interfaces.messaging.adapters: [web]) and run your agent under ast project. Open the chat at http://localhost:3100 and send a message. You should see the reply Hello, <your message>!.

Lifecycle

AgentAdapter protocol

The interface is a typing.Protocol — duck-typed. Any class with these members satisfies it.

StreamHooks

Call these from inside stream() and stream_audio(). They translate directly into outbound AgentResponse messages on the gRPC stream.

on_status_update takes a dict with a "status" key. Valid values: "THINKING", "SEARCHING", "GENERATING", "PROCESSING", "ANALYZING", "CUSTOM". For CUSTOM, include "custom_message":

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

StreamOptions

The bridge passes this to every stream() / stream_audio() call.

AudioInput

Passed to stream_audio().

VoiceProvider

Optional protocol for STT. Pass an instance to your adapter, then call its listen() method in stream_audio().

FeedbackEvent

Passed to on_feedback(). kind is a stable string discriminator — no proto imports needed.

FeedbackEvent.kind values

on_feedback may be sync or async — the bridge probes with hasattr and dispatches. It does not await the result, so don’t block on slow I/O. Push to a queue or trigger an async write and return.

Trace context

Call on_trace_context once per turn to attach the turn’s W3C trace context. The bridge stamps it on every AgentResponse it emits for the turn, so a consumer can correlate any response — and the feedback that references it — back to the trace. The hook is optional, so probe for it with getattr before calling.

create_traceparent(*, trace_id, span_id, trace_flags="01") formats a native trace/span ID pair into a W3C traceparent string. It 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))

The first-party framework adapters already emit trace context from their root span — you only wire this up in a custom adapter. See Trace context for the underlying TraceContext wire shape.

serve(adapter, options?)

Thin wrapper that handles process lifecycle. Sets up JSON-formatted logging, runs asyncio.run(bridge.start()), and exits on SIGINT / SIGTERM.

ServeOptions.server_address defaults to os.environ.get("GRPC_SERVER_ADDR", "localhost:9090").

Worked example: a plain Anthropic agent

No framework — calling Anthropic’s streaming messages API.

import os
from anthropic import AsyncAnthropic
from astropods_adapter_core import StreamHooks, StreamOptions, serve
client = AsyncAnthropic(
    api_key=os.environ["ASTRO_GATEWAY_API_KEY"],
    base_url=os.environ["ASTRO_GATEWAY_URL"],
)
class AnthropicDirect:
    name = "Anthropic Direct"
    async def stream(self, prompt: str, hooks: StreamHooks, options: StreamOptions) -> None:
        try:
            hooks.on_status_update({"status": "GENERATING"})
            async with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                system="You are a helpful assistant.",
                messages=[{"role": "user", "content": prompt}],
            ) as stream:
                async for text in stream.text_stream:
                    hooks.on_chunk(text)
            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": []}
    def on_feedback(self, event) -> None:
        # Push to your evals pipeline, Airtable, etc. Don't block.
        record_feedback_async(event)
serve(AnthropicDirect())

Worked example: audio with STT

from astropods_adapter_core import AudioInput, StreamHooks, StreamOptions, serve
class MyAdapter:
    name = "My Agent"
    def __init__(self, voice):
        self._voice = voice  # any object with: async listen(bytes, config) -> str
    async def stream(self, prompt, hooks, options):
        # ... text handling as usual
    async def stream_audio(self, audio: AudioInput, hooks: StreamHooks, options: StreamOptions) -> None:
        try:
            hooks.on_status_update({"status": "PROCESSING", "custom_message": "Transcribing audio"})
            transcript = await self._voice.listen(audio.data, audio.config)
            hooks.on_transcript(transcript)
            # Now run the agent on the transcript as if it were a text message
            await self.stream(transcript, hooks, options)
        except Exception as e:
            hooks.on_error(e)
    def get_config(self):
        return {"system_prompt": "...", "tools": []}

Rules of thumb

• Call exactly one of on_finish() or on_error() per request. Skipping either leaves the user staring at a half-rendered reply.
• Catch your own exceptions inside stream(). If you let the coroutine raise, the user sees nothing.
• Don’t block in on_feedback() — push to a queue or trigger async work and return.
• Use options.conversation_id (not platform_context.thread_id) as your memory key.
• Use only options.attachments as file input for the turn. Write output files to AGENT_FILES_DIR and call on_file before on_finish; see Files in chat.
• Configure OTEL manually before calling serve() if you want traces. Framework adapters auto-configure this; the raw core does not.

Exported symbols

from astropods_adapter_core import (
    # Protocols / interfaces
    AgentAdapter,
    StreamHooks,
    VoiceProvider,
    # Dataclasses
    StreamOptions,
    AudioInput,
    FeedbackEvent,
    ServeOptions,
    # Re-exported from astropods_messaging
    PlatformContext,
    TraceContext,
    # Tracing helper
    create_traceparent,
    # Entry points
    serve,
    MessagingBridge,
)

See Messaging SDK for the underlying proto shapes referenced by StreamHooks and PlatformContext.