Custom adapter (Python)

1
from astropods_adapter_core import StreamHooks, StreamOptions, serve
2
3
class MyAdapter:
4
    name = "My Agent"
5
6
    async def stream(self, prompt: str, hooks: StreamHooks, options: StreamOptions) -> None:
7
        try:
8
            hooks.on_chunk("Hello, ")
9
            hooks.on_chunk(f"{prompt}!")
10
            hooks.on_finish()
11
        except Exception as e:
12
            hooks.on_error(e)
13
14
    def get_config(self) -> dict:
15
        return {"system_prompt": "You are a helpful assistant.", "tools": []}
16
17
serve(MyAdapter())

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

1
import os
2
from anthropic import AsyncAnthropic
3
from astropods_adapter_core import StreamHooks, StreamOptions, serve
4
5
client = AsyncAnthropic(
6
    api_key=os.environ["ASTRO_GATEWAY_API_KEY"],
7
    base_url=os.environ["ASTRO_GATEWAY_URL"],
8
)
9
10
class AnthropicDirect:
11
    name = "Anthropic Direct"
12
13
    async def stream(self, prompt: str, hooks: StreamHooks, options: StreamOptions) -> None:
14
        try:
15
            hooks.on_status_update({"status": "GENERATING"})
16
17
            async with client.messages.stream(
18
                model="claude-sonnet-4-6",
19
                max_tokens=1024,
20
                system="You are a helpful assistant.",
21
                messages=[{"role": "user", "content": prompt}],
22
            ) as stream:
23
                async for text in stream.text_stream:
24
                    hooks.on_chunk(text)
25
26
            hooks.on_finish()
27
        except Exception as e:
28
            hooks.on_error(e)
29
30
    def get_config(self) -> dict:
31
        return {"system_prompt": "You are a helpful assistant.", "tools": []}
32
33
    def on_feedback(self, event) -> None:
34
        # Push to your evals pipeline, Airtable, etc. Don't block.
35
        record_feedback_async(event)
36
37
serve(AnthropicDirect())

1
from astropods_adapter_core import AudioInput, StreamHooks, StreamOptions, serve
2
3
class MyAdapter:
4
    name = "My Agent"
5
6
    def __init__(self, voice):
7
        self._voice = voice  # any object with: async listen(bytes, config) -> str
8
9
    async def stream(self, prompt, hooks, options):
10
        # ... text handling as usual
11
12
    async def stream_audio(self, audio: AudioInput, hooks: StreamHooks, options: StreamOptions) -> None:
13
        try:
14
            hooks.on_status_update({"status": "PROCESSING", "custom_message": "Transcribing audio"})
15
            transcript = await self._voice.listen(audio.data, audio.config)
16
            hooks.on_transcript(transcript)
17
18
            # Now run the agent on the transcript as if it were a text message
19
            await self.stream(transcript, hooks, options)
20
        except Exception as e:
21
            hooks.on_error(e)
22
23
    def get_config(self):
24
        return {"system_prompt": "...", "tools": []}

1
from astropods_adapter_core import (
2
    # Protocols / interfaces
3
    AgentAdapter,
4
    StreamHooks,
5
    VoiceProvider,
6
    # Dataclasses
7
    StreamOptions,
8
    AudioInput,
9
    FeedbackEvent,
10
    ServeOptions,
11
    # Re-exported from astropods_messaging
12
    PlatformContext,
13
    # Entry points
14
    serve,
15
    MessagingBridge,
16
)