Serving an Agent
An API for Juno.
So far, Juno has lived in a script. For a website or an app to use it, it needs to be a service, with an address. Our ML Ops course goes deep into this. Here, we'll do what matters for agents.
- Juno lives in a script
- It needs an address
- What matters for agents
An agent behind an API
FastAPI turns a function into a web service. Pydantic checks each request. Our endpoint would call the agent, and it returns a reply, plus a run ID for tracing. We use a test client, so there's no need for a real server. Run it. A good request gets a two hundred. A bad request is rejected automatically, with a four twenty two, before it reaches any model.
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel
app = FastAPI()
class Ticket(BaseModel):
customer: str
message: str
@app.post("/support")
def support(ticket: Ticket):
reply = f"Hi {ticket.customer}, I'm checking that for you."
return {"reply": reply, "run_id": "r-1042"}
client = TestClient(app)
r = client.post("/support", json={"customer": "Priya", "message": "Where is B42?"})
print(r.status_code, r.json())
print(client.post("/support", json={"customer": "Priya"}).status_code)200 {'reply': "Hi Priya, I'm checking that for you.", 'run_id': 'r-1042'}
422Packaging it
A container packages the service, so that it runs the same way everywhere. Copy the requirements first, so that builds are faster. Run as a user who isn't root. And notice what's missing. There's no key here. Secrets are passed in when it runs, and they're never baked into the image.
FROM python:3.13-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
USER nobody
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]Don't keep people waiting
Agents can take minutes, and web requests shouldn't. So accept the job at once, put it on a queue, and let workers handle it. Here, four tickets are accepted immediately, and two workers process them. Run it. Accepted first, and processed afterwards. The user gets a run ID straight away, and it can check back, or get a notification. In production, you'd use a real queue such as Redis or SQS.
import asyncio
async def worker(name, queue):
while True:
ticket = await queue.get()
await asyncio.sleep(0.1)
print(f"{name} finished {ticket}")
queue.task_done()
async def main():
queue = asyncio.Queue()
workers = [asyncio.create_task(worker(f"worker-{i}", queue)) for i in (1, 2)]
for n in range(1, 5):
await queue.put(f"ticket-{n}")
print("Accepted four tickets immediately")
await queue.join()
for w in workers:
w.cancel()
asyncio.run(main())Accepted four tickets immediately worker-1 finished ticket-1 worker-2 finished ticket-2 worker-1 finished ticket-3 worker-2 finished ticket-4
Limiting concurrency
Model providers have rate limits, and a burst can get you blocked. A semaphore caps how many calls run at once. Here, nine calls run, and never more than three together. Run it. They finish in three waves. Add that around all of your model calls, along with the backoff that you learned, and bursts stop being a problem.
import asyncio, time
async def call_model(limit, n):
async with limit:
await asyncio.sleep(0.1)
return n
async def main():
limit = asyncio.Semaphore(3)
start = time.time()
done = await asyncio.gather(*[call_model(limit, n) for n in range(9)])
print(f"{len(done)} calls, at most three at once: {time.time() - start:.1f}s")
asyncio.run(main())9 calls, at most three at once: 0.3s
Summary
An API, a container, a queue, and a rate limit. Next, the safest way to release an agent to real customers.
Key takeaways- FastAPI serves the agent
- Long work goes on a queue
- Cap concurrent model calls