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..."})

attach_file(name, data=None, mime_type=None)

Writes a file to the agent's files dir (AGENT_FILES_DIR) and attaches it to the reply. Web chat shows it as a download chip, and Slack uploads it into the thread. Call it from a tool while the agent handles a message, sync or async. It raises outside one.

from astropods_adapter_core import attach_file

attach_file("account-plan.csv", csv_text, "text/csv")

name is one path segment. Omit data to attach a file that is already in the files dir. It returns the file's name. Every user's files share that dir, so with data a name already there gets a number: summary.md becomes summary-2.md. A file over the deployment's file size limit (FILES_MAX_MB, 100 MiB when unset, 0 for no limit) raises ValueError, so the agent knows before it replies.

Reading attached files

These functions read the files a user attaches. Each takes the path from StreamOptions.attachments. The LangChain adapter wraps them as tools in file_tools().

Every thread's and user's files share AGENT_FILES_DIR, so the functions open only the files the current conversation received as attachments or wrote with attach_file, and they raise outside a turn. The bridge saves that list, up to 100 files for each conversation, in AGENT_FILES_DIR/.conversations, so it lasts through a restart. A file deleted and replaced at the same path is not on the list.

The functions deliver a file's contents and do no analysis. For counts or totals over a large table, give the agent a tool that runs pandas or DuckDB on the attachment's path, after passing it through resolve_agent_file, which raises for any file the conversation did not receive or create. pypdf, mammoth, and openpyxl load only when an agent reads a file of that kind.

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())

Connections

A connection is a third-party account, such as GitHub, that the chatting user lets the agent use as them. Install the connections extra:

pip install "astropods-adapter-core[connections]"

from astropods_adapter_core.connections import ConnectionClient, ConnectionTokenError

connections = ConnectionClient()

token = connections.get_token("github", user_id)
token = await connections.get_token_async("github", user_id)

ConnectionClient() reads ASTRO_AUTHZ_TOKEN and takes the server URL from it. The server returns a token only for a user who allowed this agent and has messaged it in web chat in the last 15 minutes. A refusal raises ConnectionTokenError with code set to not_consented, not_active, not_connected, needs_reauthorization, or unavailable.