### NPM Scripts for Development Workflow

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

Defines npm scripts for managing the development environment, including installing dependencies, running the FastAPI backend with uvicorn, starting the Next.js development server, and running both concurrently.

```json
// package.json scripts
{
  "scripts": {
    "fastapi-dev": "pip3 install -r requirements.txt && python3 -m uvicorn api.index:app --reload",
    "next-dev": "next dev",
    "dev": "concurrently \"npm run next-dev\" \"npm run fastapi-dev\"",
    "build": "next build",
    "start": "next start"
  }
}

```

--------------------------------

### Vercel Rewrites for API Proxy

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

Specifies rewrite rules for Vercel deployment to direct API requests to the appropriate backend handler. This configuration ensures that requests starting with /api/ are routed correctly.

```json
{
  "rewrites": [
    { "source": "/api/(.*)", "destination": "/api" }
  ]
}

```

--------------------------------

### Generate Server-Sent Events Stream (Python)

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

This function generates Server-Sent Events (SSE) formatted streaming responses. It supports streaming text deltas, handling tool calls (though not explicitly shown in the output example), and emitting start/finish events. It uses the OpenAI client to create a chat completion stream.

```python
import json
import uuid
from typing import Sequence, Dict, Any, Callable, Mapping
from openai import OpenAI

def stream_text(
    client: OpenAI,
    messages: Sequence[Dict[str, Any]],
    tool_definitions: Sequence[Dict[str, Any]],
    available_tools: Mapping[str, Callable],
    protocol: str = "data"
):
    def format_sse(payload: dict) -> str:
        return f"data: {json.dumps(payload, separators=(',', ':'))}\n\n"

    message_id = f"msg-{uuid.uuid4().hex}"
    yield format_sse({"type": "start", "messageId": message_id})

    stream = client.chat.completions.create(
        messages=messages,
        model="gpt-4o",
        stream=True,
        tools=tool_definitions
    )

    text_started = False
    text_stream_id = "text-1"

    for chunk in stream:
        for choice in chunk.choices:
            delta = choice.delta
            if delta and delta.content:
                if not text_started:
                    yield format_sse({"type": "text-start", "id": text_stream_id})
                    text_started = True
                yield format_sse({
                    "type": "text-delta",
                    "id": text_stream_id,
                    "delta": delta.content
                })

    if text_started:
        yield format_sse({"type": "text-end", "id": text_stream_id})

    yield format_sse({"type": "finish"})
    yield "data: [DONE]\n\n"

# Example usage with error handling:
# from openai import OpenAI
# 
# client = OpenAI(api_key="sk-...")
# messages = [{"role": "user", "content": "Hello!"}]
# 
# try:
#     for event in stream_text(client, messages, [], {}, "data"):
#         print(event, end='', flush=True)
# except Exception as e:
#     print(f"Streaming error: {e}")

```

--------------------------------

### FastAPI Middleware for Vercel Deployment

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

Python FastAPI application demonstrating middleware configuration for Vercel deployment. It includes setting headers for the Vercel SDK, handling OIDC token retrieval for authenticated API calls via the Vercel AI Gateway, and processing chat requests. Dependencies include `fastapi`, `uvicorn`, `python-dotenv`, `openai`, and `vercel`.

```python
from fastapi import FastAPI, Request as FastAPIRequest
from dotenv import load_dotenv
from vercel import oidc
from vercel.headers import set_headers

load_dotenv('.env.local')
app = FastAPI()

@app.middleware("http")
async def _vercel_set_headers(request: FastAPIRequest, call_next):
    # Set request headers for Vercel SDK
    set_headers(dict(request.headers))
    response = await call_next(request)
    return response

@app.post("/api/chat")
async def handle_chat_data(request: Request):
    # Get OIDC token for authenticated API calls
    api_key = oidc.get_vercel_oidc_token()

    # Use token with OpenAI client through Vercel AI Gateway
    client = OpenAI(
        api_key=api_key,
        base_url="https://ai-gateway.vercel.sh/v1"
    )

    # Process chat request...
    return response

# Development setup
"""
# .env.local
OPENAI_API_KEY=sk-...

# Install dependencies
pip install fastapi uvicorn python-dotenv openai vercel

# Run development server
uvicorn api.index:app --reload --port 8000

# Test the endpoint
curl -X POST http://localhost:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"Hello"}]}'


```

--------------------------------

### Define and Execute AI Function Tools (Python)

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

This code defines an AI function tool 'get_current_weather' using a specific JSON schema for its parameters. It also provides a dictionary mapping tool names to their corresponding Python functions, enabling the AI to call these tools.

```python
import requests
from typing import Dict, Any, Callable

def get_current_weather(latitude: float, longitude: float) -> Dict[str, Any]:
    url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}&current=temperature_2m&hourly=temperature_2m&daily=sunrise,sunset&timezone=auto"

    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        return {"error": str(e)}

TOOL_DEFINITIONS = [{
    "type": "function",
    "function": {
        "name": "get_current_weather",
        "description": "Get the current weather at a location",
        "parameters": {
            "type": "object",
            "properties": {
                "latitude": {
                    "type": "number",
                    "description": "The latitude of the location"
                },
                "longitude": {
                    "type": "number",
                    "description": "The longitude of the location"
                }
            },
            "required": ["latitude", "longitude"]
        }
    }
}]

AVAILABLE_TOOLS = {
    "get_current_weather": get_current_weather
}

```

--------------------------------

### FastAPI Chat Completion Streaming API (Python)

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

A FastAPI endpoint that accepts chat messages and streams AI responses using Server-Sent Events (SSE) with the Data Stream Protocol. It initializes the OpenAI client with a Vercel OIDC token and returns a StreamingResponse with specific headers for AI SDK compatibility. Input is a list of ClientMessage objects, and output is an SSE stream.

```python
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class ClientMessage(BaseModel):
    role: str
    content: str

class Request(BaseModel):
    messages: List[ClientMessage]

@app.post("/api/chat")
async def handle_chat_data(request: Request, protocol: str = Query('data')):
    # Convert messages to OpenAI format
    openai_messages = [{"role": msg.role, "content": msg.content} for msg in request.messages]

    # Initialize OpenAI client with Vercel OIDC token
    client = OpenAI(
        api_key="your-api-key",
        base_url="https://ai-gateway.vercel.sh/v1"
    )

    # Create streaming response
    response = StreamingResponse(
        stream_text(client, openai_messages, [], {}, protocol),
        media_type="text/event-stream",
    )

    # Add required headers for AI SDK streaming
    response.headers["x-vercel-ai-ui-message-stream"] = "v1"
    response.headers["Cache-Control"] = "no-cache"
    response.headers["Connection"] = "keep-alive"
    response.headers["X-Accel-Buffering"] = "no"

    return response

# Example curl request
"""
curl -X POST http://localhost:8000/api/chat \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "user", "content": "What is the weather like?"}
    ]
  }'

Expected output (SSE stream):
data: {"type":"start","messageId":"msg-abc123"}

data: {"type":"text-start","id":"text-1"}

data: {"type":"text-delta","id":"text-1","delta":"I"}

data: {"type":"text-delta","id":"text-1","delta":" can"}

data: {"type":"text-end","id":"text-1"}

data: {"type":"finish","messageMetadata":{"finishReason":"stop","usage":{"promptTokens":10,"completionTokens":20}}}

data: [DONE]
"""

```

--------------------------------

### Next.js API Route Rewrites for FastAPI Proxy

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

Configures Next.js to proxy API requests to a local FastAPI backend during development or to a serverless deployment in production. It uses environment variables to determine the correct backend endpoint.

```javascript
// next.config.js
module.exports = {
  rewrites: async () => {
    return [
      {
        source: '/api/:path*',
        destination:
          process.env.NODE_ENV === 'development'
            ? 'http://127.0.0.1:8000/api/:path*'
            : '/api/'
      }
    ];
  }
};

```

--------------------------------

### Convert Client Messages to OpenAI Format (Python)

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

A Python function that converts a list of client messages, potentially including multimodal content (text, images, tool calls), into the format expected by OpenAI's chat completion API. It handles different part types and simplifies content to a string if only a single text part exists. Input is a list of ClientMessage objects, and output is a list of dictionaries suitable for the OpenAI API.

```python
from typing import List, Optional, Any
from pydantic import BaseModel

class ClientMessagePart(BaseModel):
    type: str
    text: Optional[str] = None
    url: Optional[str] = None
    contentType: Optional[str] = None

class ClientMessage(BaseModel):
    role: str
    content: Optional[str] = None
    parts: Optional[List[ClientMessagePart]] = None

def convert_to_openai_messages(messages: List[ClientMessage]):
    openai_messages = []

    for message in messages:
        message_parts = []

        if message.parts:
            for part in message.parts:
                if part.type == 'text':
                    message_parts.append({
                        'type': 'text',
                        'text': part.text or ''
                    })
                elif part.type == 'file' and part.contentType and part.contentType.startswith('image'):
                    message_parts.append({
                        'type': 'image_url',
                        'image_url': {'url': part.url}
                    })
        elif message.content:
            message_parts.append({
                'type': 'text',
                'text': message.content
            })

        # Simplify to string if only one text part
        content = message_parts[0]['text'] if len(message_parts) == 1 and message_parts[0]['type'] == 'text' else message_parts

        openai_messages.append({
            "role": message.role,
            "content": content or ""
        })

    return openai_messages

```

--------------------------------

### React Chat Component with useChat Hook

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

Frontend React component that utilizes the `useChat` hook from the Vercel AI SDK to manage chat state and render streaming messages. It handles user input, sends messages to the backend, displays responses as they stream in, and manages loading and error states. Dependencies include `@ai-sdk/react` and `sonner` for toasts.

```typescript
'use client';

import { useChat, type UIMessage } from '@ai-sdk/react';
import { toast } from 'sonner';
import React from 'react';

export function Chat() {
  const { messages, setMessages, sendMessage, status, stop } = useChat({
    id: 'chat-001',
    onError: (error: Error) => {
      if (error.message.includes('Too many requests')) {
        toast.error('You are sending too many messages. Please try again later.');
      }
    }
  });

  const [input, setInput] = React.useState('');
  const isLoading = status === 'submitted' || status === 'streaming';

  const handleSubmit = (event?: { preventDefault?: () => void }) => {
    event?.preventDefault?.();
    if (input.trim()) {
      sendMessage({ text: input });
      setInput('');
    }
  };

  return (
    <div className="flex flex-col h-screen">
      <div className="flex-1 overflow-y-scroll">
        {messages.length === 0 && <div>Start a conversation</div>}

        {messages.map((message: UIMessage) => (
          <div key={message.id} className={`message ${message.role}`}>
            <div className="content">{message.content}</div>
          </div>
        ))}

        {isLoading && messages[messages.length - 1]?.role === 'user' && (
          <div className="thinking">AI is thinking...</div>
        )}
      </div>

      <form onSubmit={handleSubmit} className="p-4">
        <input
          type="text"
          value={input}
          onChange={(e) => setInput(e.target.value)}
          placeholder="Type a message..."
          disabled={isLoading}
          className="w-full p-2 border rounded"
        />
        <button type="submit" disabled={isLoading || !input.trim()}>
          {isLoading ? 'Stop' : 'Send'}
        </button>
      </form>
    </div>
  );
}

/*
Example usage in Next.js page:

// app/page.tsx
import { Chat } from '@/components/chat';

export default function Page() {
  return <Chat />;
}

The useChat hook automatically:
- Connects to /api/chat endpoint
- Manages message state
- Handles streaming responses
- Provides loading states
- Supports message retry and error handling

Example interaction flow:
1. User types "Hello" and submits
2. sendMessage({ text: "Hello" }) is called
3. POST request sent to /api/chat with messages array
4. Backend streams SSE events
5. useChat hook parses events and updates messages state
6. Component re-renders with new message content as it streams
*/

```

--------------------------------

### Convert Messages to OpenAI Format

Source: https://context7.com/vercel-labs/ai-sdk-preview-python-streaming/llms.txt

Converts a list of client messages, which can include text and file parts, into the format expected by OpenAI's API. This is useful for integrating with OpenAI models.

```python
from typing import Sequence, Dict, Any

# Assuming ClientMessage and ClientMessagePart are defined elsewhere
# For demonstration, let's define dummy classes if they are not provided
class ClientMessagePart:
    def __init__(self, type: str, text: str = None, contentType: str = None, url: str = None):
        self.type = type
        self.text = text
        self.contentType = contentType
        self.url = url

class ClientMessage:
    def __init__(self, role: str, content: Any = None, parts: Sequence[ClientMessagePart] = None):
        self.role = role
        self.content = content
        self.parts = parts

def convert_to_openai_messages(messages: Sequence[ClientMessage]) -> Sequence[Dict[str, Any]]:
    openai_messages = []
    for msg in messages:
        openai_msg = {"role": msg.role}
        if msg.parts:
            content_parts = []
            for part in msg.parts:
                if part.type == "text":
                    content_parts.append({"type": "text", "text": part.text})
                elif part.type == "file" and part.contentType.startswith("image"):
                    content_parts.append({"type": "image_url", "image_url": {"url": part.url}})
                # Add other part types if needed
            if len(content_parts) == 1:
                openai_msg["content"] = content_parts[0]
            else:
                openai_msg["content"] = content_parts
        elif msg.content:
            openai_msg["content"] = msg.content
        openai_messages.append(openai_msg)
    return openai_messages

# Example usage:
messages = [
    ClientMessage(role="user", parts=[
        ClientMessagePart(type="text", text="What's in this image?"),
        ClientMessagePart(type="file", contentType="image/png", url="data:image/png;base64,iVBOR...")
    ]),
    ClientMessage(role="assistant", content="This image shows a cat."),
    ClientMessage(role="user", content="Tell me more about cats.")
]

openai_messages = convert_to_openai_messages(messages)
print(openai_messages)

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.