AWS Builder Center

Builder Center

Building Neuravo: A Provider-Agnostic AI SDK on Top of AWS Bedrock's Converse API

Building Neuravo: A Provider-Agnostic AI SDK on Top of AWS Bedrock's Converse API

A small Python toolkit that lets you write your AI chat code once, then switch between AWS Bedrock and OpenAI by changing a single word — no rewriting required.

DHILIP S E

Introducing Neuravo: One Python Interface for Every LLM Provider

If you've shipped more than one LLM-backed feature, you've probably hit this wall: your Bedrock code and your OpenAI code don't look anything alike. Different auth, different response shapes, different streaming mechanics, different exceptions to catch. Swapping providers — or even just running evals against two of them side by side — means touching application code that has no business knowing which vendor answered the request.
Neuravo is a provider-agnostic Python SDK that fixes this by putting one consistent interface — one  Client , one  ChatResponse , one retry and error-handling model — in front of whichever provider is actually doing the work. This post walks through how it's built, what it gives you today, and —unusually for an SDK's public docs— exactly what it doesn't do yet.

The core idea: one contract, swappable backends

Everything starts from an abstract contract,  BaseProvider , that every provider implementation has to satisfy:
class BaseProvider(ABC):

    @abstractmethod

    async def initialize(self, config: Config) -> None: ...



    @abstractmethod

    async def chat(self, messages: List[Message]) -> ChatResponse: ...



    @abstractmethod

    async def stream_chat(self, messages: List[Message]) -> AsyncIterator[ChatResponse]: ...



    @abstractmethod

    async def health_check(self) -> HealthStatus: ...



    @abstractmethod

    async def close(self) -> None: ...
Bedrock and OpenAI both implement this exact contract. The public-facing  Client  doesn't know or care which one it's holding — it resolves a provider by name through a  ProviderRegistry  and drives it through the same five methods regardless of vendor:
config = Config(provider="bedrock", region="us-east-1")

client = Client(config)

response = await client.chat("What is machine learning?")
Switching vendors is a one-string change:
client = Client(Config(provider="openai", model="gpt-4o-mini"))
Nothing else in the calling code changes — same  chat() , same  ChatResponse , same exceptions. That's the whole pitch of the library in two lines of config.

Providers register themselves — including yours

New providers don't get special-cased into the core.  neuravo/__init__.py  imports  neuravo.providers.aws  and  neuravo.providers.openai  purely for their side effect: each module self-registers with  ProviderRegistry  on import.  providers.openai  is a no-op if you haven't installed the  openai  extra, so the dependency is genuinely optional:
pip install neuravo                # Bedrock only

pip install "neuravo[openai]"      # + OpenAI
Third-party providers plug into the exact same mechanism, but from outside the package entirely —  neuravo.plugins  discovers providers registered under a  neuravo.plugins  entry point group in any installed package, no fork of Neuravo required. Each plugin loads in isolation: one broken third-party provider raising on import doesn't take the others down with it. That's the kind of detail that only matters once someone else's badly packaged plugin is sitting in your dependency tree — and it's already handled.

Retry logic lives in exactly one place

A subtle but important design choice: retry/backoff isn't duplicated per provider.  Client.chat  wraps every provider call in a shared  ExponentialBackoffRetry , configured straight from  Config :
retry = ExponentialBackoffRetry(

    RetryConfig(

        max_retries=self.config.max_retries,

        backoff_factor=self.config.backoff_factor,

    )

)

response = await retry.execute(self._provider.chat, self._history.get_all())
Providers only need to implement the happy path plus honest failures — they raise a normalized  ProviderError  and let the client's retry policy decide what to do about it. That's a meaningful chunk of complexity that individual provider implementations, including any third-party plugin, get for free.
Config validation follows the same "fail fast, fail clearly" philosophy — it's a frozen Pydantic model, so a  Config  instance can't be mutated after construction, and bad values ( max_retries  outside  0-10 ,  backoff_factor  outside  1.0-10.0 , an empty provider string) are rejected at construction time with a clear  ValueError , not three call stacks deep during a live request.

Streaming, the same way, everywhere

Streaming follows the identical pattern — an abstract  stream_chat  on  BaseProvider , wrapped by  Client.stream , which accumulates the full response into conversation history as chunks arrive:
async for chunk in client.stream("Tell me a story"):

    print(chunk.content, end="", flush=True)
One implementation detail worth calling out for anyone implementing a new provider: the abstract  stream_chat / stream  methods contain an unreachable  yield  behind  if False . It looks pointless until you remember that Python decides whether a function is a generator at parse time — without that yield, mypy infers a plain coroutine, and every real subclass's  AsyncIterator  return type would mismatch the base class. Small, easy to trip over, already solved.

What building the Bedrock provider actually required

The clean five-method contract above hides three problems that only surface once you try to satisfy it against a real vendor SDK — not against a mock.
Wire formats don't agree. Bedrock's Converse API wants system instructions pulled into their own  system  field, separate from the conversation turns:
# ChatHandler.format_messages (Bedrock)

for message in messages:

    if message.role == "system":

        system_blocks.append({"text": message.content})

    else:

        turns.append({"role": message.role, "content": [{"text": message.content}]})

return turns, system_blocks
OpenAI, by contrast, accepts  system  as a normal message role inline — its formatter's docstring says so directly: "Unlike Bedrock's Converse API, OpenAI accepts  system  as a normal message role, so no splitting is needed." Each provider owns exactly one function that knows its vendor's wire format; nothing above that layer —  Client , retry, conversation history — ever branches on which one it's talking to.
 boto3  can't do async. The library Bedrock runs on is synchronous.  converse_stream 's event iterator performs a blocking network read on every call — run it directly inside an  async def  and it stalls the whole event loop, not just the current request, until AWS finishes talking.  StreamingHandler.handle_stream  works around this by reading events on a background thread and handing them to the async side through a  queue.Queue :
def _consume() -> None:

    for event in stream_response["stream"]:   # blocking iterator

        q.put(event)

    q.put(_STREAM_DONE)



threading.Thread(target=_consume, daemon=True).start()



while True:

    event = await loop.run_in_executor(None, q.get)   # non-blocking wait

    if event is _STREAM_DONE:

        break

    ...
The async generator that  Client.stream  consumes never touches  boto3  directly — only the queue.
One Bedrock client isn't enough. The client used to send messages ( bedrock-runtime ) has no method to list models or report service health —  list_foundation_models  only exists on a second, separate client ( bedrock ). Calling the wrong client for a health check fails in a way that looks like a permissions problem rather than a missing method, until you go looking.  BedrockProvider.initialize  creates both up front rather than forcing one client to do both jobs:
self.client = await asyncio.to_thread(boto3.client, "bedrock-runtime", **client_kwargs)

# Separate control-plane client (model listing, health checks) -

# "bedrock-runtime" only exposes converse/converse_stream/invoke_model.

self._control_client = await asyncio.to_thread(boto3.client, "bedrock", **client_kwargs)
 chat() / stream_chat()  use  self.client ;  health_check()  calls  self._control_client.list_foundation_models() . None of this is visible from  Client  — it's entirely contained inside the one provider that has to deal with it.

Observability, cost, evaluation, and everything past the MVP

Past the core chat/streaming loop, Neuravo ships a full production toolbox, all built on the same "one interface, works for any provider" philosophy:
  •  observability/  — a  Tracer  with nested spans, a  MetricsRegistry  (counters, gauges, histograms), an in-process  Monitor , and a  CloudWatchExporter  for shipping metrics off-box.
  •  cost/  — a per-model pricing table plus a  CostTracker  that turns token usage into running-total USD cost, broken down per model.
  •  evaluation/  —  PromptTemplate -driven  EvalCase s run through an  EvaluationRunner  and scored with pluggable scorers ( exact_match ,  contains ,  regex_match ).
  •  prompts/  — named, versioned prompt templates via  PromptRegistry .
  •  benchmark/  —  run_benchmark  computes latency stats (mean, percentiles) over repeated calls against any client, any provider.
  •  security/  — a  SecurityPolicy  combining block-list prompt checks with output redaction; sensitive values (API keys, tokens, credentials) are also automatically stripped from log output.
  •  workflows/  — a sequential  Step / Workflow  pipeline that threads a shared context through multi-step operations.
Every one of these is real, tested code on  main  today, not a slide-deck promise.

The part most SDKs don't tell you: what's still a stub

This is the detail that actually made this codebase worth writing about. Most project READMEs are aspirational — a feature list of what the maintainers intend to ship, blurred together with what already works. Neuravo's   docs/roadmap.md   draws that line explicitly, module by module:
Module
Status
 cache/ 
 Cache.get / set / clear  exist with the right signatures — every body is an unimplemented  pass . No caching happens.
 retry/rate_limit.py 
 RateLimiter / RateLimitConfig  have a token-bucket shape, but  acquire()  and  _refill()  are  pass . Nothing is enforced.
 retry/timeout.py 
 TimeoutManager / TimeoutConfig  exist;  execute_with_timeout()  is  pass . (The retry/backoff that  Client.chat  actually uses is a separate, fully implemented path — this stub is specifically about enforced per-call timeouts.)
 agents/ ,  embeddings/ ,  memory/ 
Empty packages — an  __init__.py  and nothing else. Reserved namespace, no committed design.
If you're evaluating Neuravo for a use case that leans on caching, rate limiting, enforced timeouts, agents, embeddings, or memory, the roadmap tells you outright: assume none of it exists yet. That's a small thing for a maintainer to write down and an easy thing to skip — most projects would rather let an empty  agents/  folder imply momentum than admit it's just a placeholder. Neuravo does the opposite, and it's a better signal of engineering discipline than any feature list would be.

Why the shape of this matters

The interesting engineering decision in Neuravo isn't any single module — it's that the boundary between "core contract" and "everything else" is drawn so cleanly that retry, streaming, observability, cost tracking, and plugin loading all sit outside any individual provider. A new provider — official or third-party — only has to implement five methods on  BaseProvider  and it inherits retry policy, history management, and the whole observability stack for free. That's the actual payoff of the "provider-agnostic" pitch: not just that Bedrock and OpenAI look the same from the outside, but that adding a third provider is cheap precisely because so little logic lives inside providers to begin with.

Try it

pip install neuravo
import asyncio

from neuravo import Client, Config



async def main():

    client = Client(Config(provider="bedrock", region="us-east-1"))

    try:

        response = await client.chat("What is machine learning?")

        print(response.content)

    finally:

        await client.close()



asyncio.run(main())
More runnable examples live in   examples/  , and the full module-by-module status lives in   docs/roadmap.md  .
Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.
Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article