Ensure the baseline example app works

Author: anbuzin

examples/fastapi-vite/backend/__init__.py

@@ -1,30 +1,80 @@
 """FastAPI application entry point."""
 
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+
+import agent
 import fastapi
 import fastapi.middleware.cors
-from routes import chat
+import fastapi.responses
+import pydantic
+import storage
 
-api = fastapi.FastAPI(
+import vercel_ai_sdk as ai
+import vercel_ai_sdk.ai_sdk_ui
+
+app = fastapi.FastAPI(
     title="py-ai-fastapi-chat",
     description="Chat demo using Python Vercel AI SDK",
 )
 
-api.add_middleware(
+app.add_middleware(
     fastapi.middleware.cors.CORSMiddleware,
     allow_origins=["*"],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-api.include_router(chat.router)
-
 
-@api.get("/health")
+@app.get("/health")
 async def health() -> dict[str, str]:
     """Health check endpoint."""
     return {"status": "ok"}
 
 
-app = fastapi.FastAPI()
-app.mount("/api", api)
+file_storage = storage.FileStorage()
+
+
+class ChatRequest(pydantic.BaseModel):
+    """Request body for the chat endpoint."""
+
+    messages: list[ai.ai_sdk_ui.UIMessage]
+    session_id: str | None = None
+
+
+@app.post("/chat")
+async def chat(request: ChatRequest) -> fastapi.responses.StreamingResponse:
+    """Handle chat requests and stream responses."""
+    messages = ai.ai_sdk_ui.to_messages(request.messages)
+    session_id = request.session_id or "default"
+    checkpoint_key = f"checkpoint:{session_id}"
+
+    llm = agent.get_llm()
+
+    # Checkpoints resume an *interrupted* run (e.g. a hook that needed
+    # user input in serverless mode).  Each normal chat turn is a fresh
+    # run — the frontend carries the full message history — so we only
+    # load a checkpoint when one was saved from a previous incomplete run.
+    saved = await file_storage.get(checkpoint_key)
+    checkpoint = ai.Checkpoint.model_validate(saved) if saved else None
+
+    result = ai.run(agent.graph, llm, messages, agent.TOOLS, checkpoint=checkpoint)
+
+    async def stream_response() -> AsyncGenerator[str]:
+        async for chunk in ai.ai_sdk_ui.to_sse_stream(result):
+            yield chunk
+
+        # If the run completed (no pending hooks), clear the checkpoint
+        # so the next request starts fresh.  If hooks are pending, save
+        # the checkpoint so the next request can resume from here.
+        if result.pending_hooks:
+            await file_storage.put(checkpoint_key, result.checkpoint.model_dump())
+        else:
+            await file_storage.delete(checkpoint_key)
+
+    return fastapi.responses.StreamingResponse(
+        stream_response(),
+        headers=ai.ai_sdk_ui.UI_MESSAGE_STREAM_HEADERS,
+    )

examples/fastapi-vite/backend/main.py

@@ -5,5 +5,9 @@ description = "Chat demo using Python Vercel AI SDK with FastAPI"
 requires-python = ">=3.12"
 dependencies = [
     "fastapi[standard]>=0.128.1",
-    "vercel-ai-sdk>=0.0.1.dev5",
+    "vercel-ai-sdk",
+    # "vercel-ai-sdk>=0.0.1.dev5",
 ]
+
+[tool.uv.sources]
+vercel-ai-sdk = { path = "../../.." }

examples/fastapi-vite/backend/pyproject.toml

@@ -11,9 +11,4 @@ export default defineConfig({
       '@': path.resolve(__dirname, './src'),
     },
   },
-  server: {
-    proxy: {
-      '/api': 'http://localhost:8000',
-    },
-  },
 })