Examples
FastAPI Integration
Use AsyncSolwyn in a FastAPI application with lifespan management
import os
from contextlib import asynccontextmanager
from openai import AsyncOpenAI
from fastapi import FastAPI
from solwyn import AsyncSolwyn
@asynccontextmanager
async def lifespan(app: FastAPI):
# Start: create and store the Solwyn client
app.state.llm = AsyncSolwyn(
AsyncOpenAI(),
api_key=os.environ["SOLWYN_API_KEY"],
)
async with app.state.llm:
yield
# Shutdown: close() is called by the async context manager
app = FastAPI(lifespan=lifespan)
@app.post("/chat")
async def chat(message: str):
response = await app.state.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": message}],
)
return {"reply": response.choices[0].message.content}This example shows how to integrate AsyncSolwyn into a FastAPI application using the lifespan pattern for proper startup and shutdown.
Why lifespan?
FastAPI's lifespan pattern is the recommended way to manage resources that should be created once at startup and cleaned up at shutdown. AsyncSolwyn fits this pattern because:
- The client should be created once and shared across requests (not per-request).
- The background metadata reporter needs to be started with
__aenter__and stopped with__aexit__. - HTTP connections should be properly closed on shutdown.
Complete example
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from openai import AsyncOpenAI
from pydantic import BaseModel
from solwyn import AsyncSolwyn, BudgetExceededError
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.llm = AsyncSolwyn(
AsyncOpenAI(),
api_key=os.environ["SOLWYN_API_KEY"],
budget_mode="hard_deny",
)
async with app.state.llm:
yield
app = FastAPI(lifespan=lifespan)
class ChatRequest(BaseModel):
message: str
model: str = "gpt-4o"
class ChatResponse(BaseModel):
reply: str
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
try:
response = await app.state.llm.chat.completions.create(
model=request.model,
messages=[{"role": "user", "content": request.message}],
)
return ChatResponse(reply=response.choices[0].message.content)
except BudgetExceededError as e:
raise HTTPException(
status_code=429,
detail=f"Budget exceeded: ${e.current_usage:.2f} of ${e.budget_limit:.2f} used",
)Multiple agents in FastAPI
Use separate AsyncSolwyn instances for different agent roles:
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from openai import AsyncOpenAI
from solwyn import AsyncSolwyn
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.fast_llm = AsyncSolwyn(
AsyncOpenAI(),
api_key=os.environ["SOLWYN_FAST_API_KEY"],
)
app.state.smart_llm = AsyncSolwyn(
AsyncOpenAI(),
api_key=os.environ["SOLWYN_SMART_API_KEY"],
)
async with app.state.fast_llm, app.state.smart_llm:
yield
app = FastAPI(lifespan=lifespan)
@app.post("/quick-answer")
async def quick_answer(message: str):
response = await app.state.fast_llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": message}],
)
return {"reply": response.choices[0].message.content}
@app.post("/deep-analysis")
async def deep_analysis(message: str):
response = await app.state.smart_llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": message}],
)
return {"reply": response.choices[0].message.content}Each endpoint's costs are tracked under its own project ID in the Solwyn dashboard.
Dependency injection alternative
If you prefer FastAPI's dependency injection over app.state:
import os
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI
from openai import AsyncOpenAI
from solwyn import AsyncSolwyn
_llm_client: AsyncSolwyn | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global _llm_client
_llm_client = AsyncSolwyn(
AsyncOpenAI(),
api_key=os.environ["SOLWYN_API_KEY"],
)
async with _llm_client:
yield
_llm_client = None
app = FastAPI(lifespan=lifespan)
def get_llm() -> AsyncSolwyn:
assert _llm_client is not None
return _llm_client
LLM = Annotated[AsyncSolwyn, Depends(get_llm)]
@app.post("/chat")
async def chat(llm: LLM, message: str):
response = await llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": message}],
)
return {"reply": response.choices[0].message.content}