How to Test Ad Injection Locally in FastAPI

Overview

You can develop ad injection for a FastAPI app locally before going to production. Point the SDK at a local ad-server address and turn SSL off; everything else — the injection call, session context, fallback — behaves the same as production.

This lets you verify that your FastAPI stream flows through injection and reaches the client correctly without needing production credentials wired up.

FastAPI apps are usually built for LLM API backends, SSE/streaming chat endpoints, and agent microservices, so test ad injection locally typically comes up while a user is mid-conversation, the moment where monetization has to be additive rather than disruptive.

How this works on FastAPI

There is no official Monetzly Python SDK — the package is Node-only. The supported path for a Python stack is the gRPC service the package ships as tps_alter.proto: generate Python stubs and open a ProcessStream bidi call (StartRequest, then a TokenRequest per LLM token, then StopRequest), reading back the ad-injected TokenResponses. Alternatively, run a tiny Node sidecar that uses the SDK and stream tokens to it. Either way, confirm the auth handshake for non-JS clients before publishing.

Since FastAPI is a Python stack with no official SDK, this task runs through the shipped tps_alter.proto gRPC service (or a Node sidecar); the auth handshake for a direct client is still TODO_VERIFY.

Steps

1. Set the server address to your local host (e.g. localhost:8080) and disable SSL.
2. Run your FastAPI app and send a test prompt through the injection path.
3. Confirm tokens stream to the client, then switch to your production host and SSL for release.

Integration snippet

# 1) Generate stubs from the proto shipped in the SDK package:
#    python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. tps_alter.proto
import grpc
import tps_alter_pb2 as pb
import tps_alter_pb2_grpc as rpc

async def inject(llm_tokens, prompt, session_id, api_key):
    # TODO_VERIFY: how is the API key passed for a direct gRPC client?
    # (metadata key name / channel credentials). Confirm with Monetzly.
    creds = grpc.ssl_channel_credentials()
    async with grpc.aio.secure_channel("your-server.com:443", creds) as ch:
        stub = rpc.TPSAlterServiceStub(ch)

        async def requests():
            yield pb.StreamRequest(start=pb.StartRequest(
                prompt=prompt, session_id=session_id, metadata={"api_key": api_key}))
            async for tok in llm_tokens:
                yield pb.StreamRequest(token=pb.TokenRequest(token=tok))
            yield pb.StreamRequest(stop=pb.StopRequest(session_id=session_id))

        async for resp in stub.ProcessStream(requests()):
            if resp.HasField("token"):
                yield resp.token.token  # ad-injected token

# In a FastAPI route, feed your LLM's token generator into inject(...) and
# return the results with StreamingResponse.

Frequently asked questions

Related