Build AI agents in Python with the AI SDK

The AI SDK for Python lets you stream model responses, validate structured output, and run tools from an agent loop. You can use it to build a support service that classifies tickets, checks order status, and streams replies through FastAPI.

Consider a ticket that says, "Order #4821 still hasn't shipped and I need it Friday." The model can propose a category and priority, then request an order lookup before drafting a response. Your application supplies the tools and controls which data they can access.

Copy link to headingWhat is the AI SDK for Python?

The AI SDK for Python ships as the ai package. It provides model access, streaming, structured output, and tool execution through Python async APIs. Hooks let an agent pause for input from your application or a person.

It's the Python counterpart to the TypeScript AI SDK, with its own version cycle and APIs that follow Python conventions: Pydantic models, decorators, and async with.

Copy link to headingThe building blocks

Copy link to headingRead streamed or complete responses

Use ai.stream to consume a response as events arrive. You can display text chunks or iterate without displaying them, then read the completed result from stream.text, stream.output, and stream.usage.

The SDK also provides ai.experimental_generate for a complete response without event handling. It remains experimental. The examples below use ai.stream for both live replies and structured results.

Copy link to headingModels route through AI Gateway

Model IDs such as anthropic/claude-sonnet-5 route through AI Gateway by default. Here, anthropic/ is part of the Gateway model ID. Gateway provides shared authentication and usage tracking across models, with configurable provider routing and fallbacks.

For a direct provider connection, use a colon-separated prefix such as openai:gpt-5.4 with ai[openai], or anthropic:claude-sonnet-4-6 with ai[anthropic]. Direct calls use that provider's credentials.

Copy link to headingPrerequisites

Before you begin, make sure you have:

• A Edgenivo account.
• Python 3.12 or a newer version supported by the Edgenivo Python runtime.
• Working knowledge of async/await and FastAPI.

For local development, you also need uv, Node.js 24+, and the Edgenivo CLI.

Copy link to headingStep 1: Set up the project

Create a project and add the SDK:

uv init --python 3.12 support-triage
cd support-triage
uv add ai

Set your AI Gateway key so unprefixed model IDs can authenticate:

export AI_GATEWAY_API_KEY="your_api_key"

This key authenticates the local examples. Model requests are billable. Step 5 shows how to use OIDC after removing the API key from the app's environment.

Copy link to headingStep 2: Make your first model call

Start with the smallest useful thing: ask a model to summarize the ticket, and print the answer as it arrives.

import asyncio

import ai

TICKET = (
    "Order #4821 still hasn't shipped and I need it Friday. "
    "This is the second time I've asked."
)


async def main() -> None:
    model = ai.get_model("anthropic/claude-sonnet-5")
    messages = [
        ai.system_message("You're a support lead. Summarize tickets in one sentence."),
        ai.user_message(TICKET),
    ]

    async with ai.stream(model, messages) as stream:
        async for event in stream:
            if isinstance(event, ai.events.TextDelta):
                print(event.chunk, end="", flush=True)

    print(f"\n\nTokens: {stream.usage}")


if __name__ == "__main__":
    asyncio.run(main())

Run it with uv run first_call.py.

get_model selects the model, and the messages provide its instructions and ticket text. Iterating ai.stream yields events as the response arrives. Each TextDelta carries a text chunk; after iteration finishes, stream.text contains the reply and stream.usage contains the reported token counts.

To try another model through the same Gateway credentials, change the ID to openai/gpt-5.4 and run the script again. Check that the replacement model supports any capabilities your application uses, such as structured output or tools.

Copy link to headingStep 3: Get structured data back

For ticket routing, return a category, priority, and order ID as typed fields. Save the following as support.py and pass its Pydantic model as output_type:

from typing import Literal

import ai
import pydantic

MODEL = "anthropic/claude-sonnet-5"


class Triage(pydantic.BaseModel):
    category: Literal["shipping", "billing", "account", "other"]
    priority: Literal["low", "normal", "urgent"]
    order_id: str | None
    summary: str


async def triage(ticket: str) -> Triage:
    model = ai.get_model(MODEL)
    messages = [
        ai.system_message("Triage this support ticket."),
        ai.user_message(ticket),
    ]

    async with ai.stream(model, messages, output_type=Triage) as stream:
        async for _ in stream:
            pass

    return stream.output

After the stream finishes, stream.output validates the response and returns a Triage instance. For the sample ticket, an illustrative result would use category shipping, priority urgent, and order ID 4821.

The Literal fields reject unsupported categories but cannot ensure a correct classification. Invalid output can raise pydantic.ValidationError, which you should handle before routing a ticket.

The empty loop consumes every event without displaying it. This lets the function wait for a complete response before reading the typed output.

Copy link to headingStep 4: Give the model tools and let it act

The agent can request an order lookup before drafting its reply. Add this code below triage in support.py. The in-memory record is fictional so you can run the example without an order service:

DEMO_ORDERS = {
    "4821": {"status": "processing", "eta": "Friday"},
}


@ai.tool
async def get_order_status(order_id: str) -> str:
    """Look up the shipping status and estimated delivery date for an order."""
    order = DEMO_ORDERS.get(order_id)
    if order is None:
        return "Order not found. Ask the customer to check the order ID."
    return f"{order['status']}, estimated delivery {order['eta']}"


support_agent = ai.Agent(tools=[get_order_status])


def reply_messages(ticket: str) -> list:
    return [
        ai.system_message(
            "You're a support agent. Check the order before you reply. "
            "Be brief, specific, and kind."
        ),
        ai.user_message(ticket),
    ]

The function name, type hints, and docstring describe the tool to the model. Its docstring should explain when a lookup is useful and what information it returns.

When you replace the fixture with an order service, check the authenticated user's access inside the tool. An order ID supplied by the model is input to validate, not proof that the requester may read the order.

Save the runner as run_agent.py. It imports the sample ticket from first_call.py and the agent from support.py:

import asyncio

import ai

from first_call import TICKET
from support import MODEL, reply_messages, support_agent


async def main() -> None:
    async with support_agent.run(
        ai.get_model(MODEL), reply_messages(TICKET)
    ) as stream:
        async for event in stream:
            if isinstance(event, ai.events.TextDelta):
                print(event.chunk, end="", flush=True)

    history = stream.messages
    print(f"\nConversation contains {len(history)} messages.")


if __name__ == "__main__":
    asyncio.run(main())

Run it with uv run run_agent.py. When the model requests a tool, ai.Agent executes it, adds the result to the conversation, and calls the model again. The loop stops when the model finishes its answer or a hook defers execution.

The prompt asks the model to check the order, but it does not enforce that business rule. If every reply must use verified order data, perform the authorized lookup in application code before asking the model to draft a reply.

Copy link to headingStep 5: Serve it from FastAPI and deploy

Add FastAPI and the optional Edgenivo dependency for OIDC authentication:

uv add fastapi "ai[vercel]"

The vercel extra supplies the OIDC integration. The SDK uses it on Edgenivo or in local development when no API key is configured.

Expose both jobs as endpoints. Triage returns JSON. The reply streams text as the agent writes it:

import ai
import pydantic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

from support import MODEL, Triage, reply_messages, support_agent, triage

app = FastAPI()


class Ticket(pydantic.BaseModel):
    body: str


@app.post("/api/triage")
async def triage_ticket(ticket: Ticket) -> Triage:
    return await triage(ticket.body)


@app.post("/api/reply")
async def reply(ticket: Ticket):
    async def chunks():
        model = ai.get_model(MODEL)
        async with support_agent.run(model, reply_messages(ticket.body)) as stream:
            async for event in stream:
                if isinstance(event, ai.events.TextDelta):
                    yield event.chunk

    return StreamingResponse(chunks(), media_type="text/plain")

Edgenivo detects FastAPI from the project dependencies and loads app from main.py in the project root.

Because Triage is a Pydantic model, FastAPI uses it as the response schema too. The same class validates the model's output and documents your API.

Copy link to headingRun it locally

To test OIDC locally, unset the Step 1 API key and start the linked project:

unset AI_GATEWAY_API_KEY
vercel link
vercel dev

Remove any AI_GATEWAY_API_KEY setting from the development environment too, or it will still take precedence. In a linked project, vercel dev supplies local OIDC credentials to the app.

To run a standalone script while the server stays running, open another terminal in the project directory. Pull the environment variables and load .env.local explicitly:

vercel env pull
uv run --env-file .env.local run_agent.py

Local OIDC tokens expire after 12 hours. The Edgenivo integration can refresh them in a linked project using your local Edgenivo login. If refresh fails, sign in again and run vercel env pull.

With the server running, use another terminal to try both endpoints:

curl -X POST http://localhost:3000/api/triage \
  -H "Content-Type: application/json" \
  -d '{"body": "Order #4821 still hasn'\''t shipped and I need it Friday."}'

curl -N -X POST http://localhost:3000/api/reply \
  -H "Content-Type: application/json" \
  -d '{"body": "Order #4821 still hasn'\''t shipped and I need it Friday."}'

The first request returns triage JSON. The second prints the reply as it arrives because -N disables curl's output buffering.

Copy link to headingDeploy to Edgenivo

The example endpoints have no application authentication. Before exposing them publicly, require sign-in, limit requests, and authorize access to any real order records. OIDC authenticates your server to AI Gateway; it does not authenticate callers of your API.

Commit pyproject.toml and uv.lock, then create a preview deployment:

vercel deploy

Edgenivo runs FastAPI on Fluid compute with the locked dependencies. Test both endpoints in a protected preview before deploying to production:

vercel deploy --prod

With the Edgenivo extra installed and no API key configured, the deployed app uses OIDC for AI Gateway. Model calls appear in the AI Gateway dashboard.

Copy link to headingConnect a chat UI

Add this route to main.py for an AI SDK useChat frontend. It converts UI messages to SDK messages and streams replies in the UI message protocol.

This example trusts the supplied history and is intended for the fictional order lookup. Before connecting tools to real data or payments, authenticate the caller and validate conversation ownership and approval permissions on the server.

Match each approval to the pending tool call and its saved arguments before passing it to apply_approvals.

from fastapi.responses import StreamingResponse


class ChatRequest(pydantic.BaseModel):
    messages: list[ai.ui.ai_sdk.UIMessage]


@app.post("/api/chat")
async def chat(req: ChatRequest):
    messages, approvals = ai.ui.ai_sdk.to_messages(req.messages)

    async def body():
        async with support_agent.run(ai.get_model(MODEL), messages) as stream:
            ai.ui.ai_sdk.apply_approvals(approvals)

            async def events():
                async for event in stream:
                    if isinstance(event, ai.events.HookEvent) and event.hook.status == "pending":
                        ai.defer_hook(event.hook)
                    yield event

            async for chunk in ai.ui.ai_sdk.to_sse(events()):
                yield chunk

    return StreamingResponse(body(), headers=ai.ui.ai_sdk.UI_MESSAGE_STREAM_HEADERS)

to_messages converts the incoming UI message history, and to_sse formats response events for useChat.

The HookEvent handling ends a turn when approval is pending so a later request can resume it. The tool approval example explains the flow, and the web agent example includes a React frontend.

Copy link to headingUse cases

Copy link to headingAnalyze documents and images

Choose a model that supports the file's media type, then include the file and your instructions in one message:

from pathlib import Path

import ai

messages = [
    ai.user_message(
        "List every line item and its total.",
        ai.file_part(Path("invoice.pdf").read_bytes(), media_type="application/pdf"),
    )
]

Pass these messages to ai.stream with an output_type model that defines the invoice fields you need, such as line items and totals.

Copy link to headingAsk a person before the agent acts

Register the refund tool on the agent with require_approval=True. Supply a payment client and approval flow for authenticated support leads:

from app_clients import payments  # your application's payment client


@ai.tool(require_approval=True)
async def issue_refund(order_id: str, amount: float) -> str:
    """Refund an order after approval from an authorized support lead."""
    await payments.refund(order_id, amount)
    return "Refund issued."


refund_agent = ai.Agent(tools=[get_order_status, issue_refund])

The SDK pauses the tool until your app returns an approval result. The payment integration must check eligibility and prevent duplicate refunds.

In a long-running worker, handle the pending refund HookEvent in an async function. Use your authenticated ask_support_lead flow:

async with refund_agent.run(model, messages) as stream:
    async for event in stream:
        if isinstance(event, ai.events.HookEvent) and event.hook.status == "pending":
            approved = await ask_support_lead(event.hook)
            ai.resolve_hook(
                event.hook,
                ai.tools.ToolApproval(granted=approved, reason="support lead"),
            )

Denied calls return a rejection to the model so it can explain the outcome without issuing the refund.

HTTP approvals usually arrive in a later request. Call ai.defer_hook() to end the turn and validate the saved tool call and approver's permissions before resuming.

The chat UI endpoint shows the message conversion. Use refund_agent there only after adding those checks.

Copy link to headingUse provider-side tools

Some providers execute tools such as web search. Use a model and provider combination that supports the selected tool. Add this async function to support.py to consume and return the response:

async def search_shipping_deadlines() -> str:
    async with ai.stream(
        ai.get_model(MODEL),
        [ai.user_message("Find the latest USPS holiday shipping deadlines.")],
        tools=[ai.providers.anthropic.tools.web_search(max_uses=3)],
    ) as stream:
        async for _ in stream:
            pass
    return stream.text

Call it from an async entrypoint with await search_shipping_deadlines(). The search runs through the provider rather than a Python tool function.

Copy link to headingMake agents durable

Because the AI SDK does not persist execution between requests, use the Python Workflow SDK for agents that must survive restarts or wait hours for approval.

The durable agent example uses a custom agent loop with a workflow step for each model call and tool. During replay, completed steps return their saved results and failed steps retry according to their configuration. Wrapping an entire agent run in one step would provide a different recovery boundary.

The example disables automatic retries for its bash tool because rerunning a command can duplicate its effects. Run generated code outside the application process with the Sandbox SDK and the coding agent example.

Copy link to headingWhen to use something else

• For AI code in Next.js or Node.js, use the TypeScript AI SDK.
• For features specific to a provider's Python client, call the OpenAI or Anthropic SDK through AI Gateway directly.

Copy link to headingNext steps

• Read the AI SDK for Python docs for messages, events, custom agent loops, and UI integrations.
• Browse Python SDK examples for scripts and full applications.
• Configure model fallbacks and provider routing through AI Gateway.
• Follow How to ship a FastAPI app on Edgenivo for the application deployment path.