### Quick Start: Initialize MCPServer and Add OpenAPI Spec

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/openapi.md

Initialize the MCPServer and add an OpenAPI specification from a URL with authentication. This is the quickest way to get started.

```python
from ai_infra import MCPServer

server = MCPServer()
server.add_openapi(
    "https://api.example.com/openapi.json",
    auth={"Authorization": "Bearer token123"}
)
server.run()
```

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

### Development Environment Setup

Source: https://github.com/nfraxlab/ai-infra/blob/main/CONTRIBUTING.md

Commands to clone the repository, install dependencies, and run quality assurance checks.

```bash
git clone https://github.com/<your-username>/ai-infra.git
cd ai-infra

# Add upstream remote
git remote add upstream https://github.com/nfraxlab/ai-infra.git

# Install dependencies
poetry install

# Activate the virtual environment
poetry shell

# Run tests
pytest -q

# Run linting
ruff check

# Run type checking
mypy src
```

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

### Quick Start: Create and Run an MCP Server

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/server.md

Initialize an MCPServer, add tools from Python functions, and run the server. Ensure ai-infra is installed.

```python
from ai_infra import MCPServer, mcp_from_functions

def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: 72°F"

server = MCPServer()
server.add_tools(mcp_from_functions([get_weather]))
server.run()
```

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

### Quick Start Graph Workflow

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/core/graph.md

A complete example demonstrating state definition, node creation, conditional branching, and execution.

```python
from ai_infra import Graph
from typing import TypedDict

class State(TypedDict):
    messages: list
    count: int

def increment(state: State) -> State:
    return {"count": state["count"] + 1}

def check_done(state: State) -> str:
    return "done" if state["count"] >= 3 else "continue"

graph = Graph(State)
graph.add_node("increment", increment)
graph.add_conditional_edges("increment", check_done, {
    "continue": "increment",
    "done": "__end__"
})
graph.set_entry_point("increment")

result = graph.run({"messages": [], "count": 0})
print(result["count"])  # 3
```

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

### Quick Start Streaming Agent Responses

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/streaming.md

Get streaming working in 3 lines. This example shows basic streaming by printing tokens as they arrive.

```python
from ai_infra import Agent

agent = Agent(tools=[my_tool])

# Basic streaming - print tokens as they arrive
async for event in agent.astream("What is the weather in NYC?"):
    if event.type == "token":
        print(event.content, end="", flush=True)
```

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

### TTS Quick Start

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/tts.md

A basic example demonstrating how to initialize the TTS client and convert text to speech.

```APIDOC
## Quick Start

```python
from ai_infra import TTS

tts = TTS()
audio = tts.speak("Hello, world!")
audio.save("output.mp3")
```
```

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

### Full Example: MCPServer with GitHub API

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/openapi.md

A comprehensive example initializing MCPServer with specific configurations and adding the GitHub API with filtering and a prefix.

```python
from ai_infra import MCPServer

# Create server
server = MCPServer(
    name="api-gateway",
    version="1.0.0",
    description="MCP gateway for external APIs"
)

# Add GitHub API
server.add_openapi(
    "https://raw.githubusercontent.com/github/rest-api-description/main/descriptions/api.github.com/api.github.com.json",
    auth={"Authorization": "Bearer ghp_xxx"},
    include=["repos/list-for-authenticated-user", "repos/get"],
    prefix="github_"
)
```

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

### Run Example with API Key

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Execute a Python example script, providing an API key as an environment variable for authentication.

```bash
OPENAI_API_KEY=sk-... python chat/01_basic.py
```

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

### Run a Single Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Execute a specific Python example script from the command line.

```bash
python chat/01_basic.py
```

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

### Run Basic Agent Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/README.md

Execute a basic agent example script. This command runs a pre-defined agent script located in the examples directory.

```bash
poetry run python examples/agents/01_basic_tools.py
```

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

### Full Example: AI Assistant Tools Server

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/server.md

A comprehensive example demonstrating the creation of an MCPServer with multiple tools, including functions for weather, web search, and calculation, using Pydantic for structured parameters.

```python
from ai_infra import MCPServer, mcp_from_functions
from pydantic import BaseModel, Field

# Define tools
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # In reality, call a weather API
    return f"Weather in {city}: 72°F, sunny"

def search_web(query: str, limit: int = 5) -> list[str]:
    """Search the web for information."""
    # In reality, call a search API
    return [f"Result {i} for: {query}" for i in range(limit)]

class CalculateParams(BaseModel):
    expression: str = Field(description="Math expression to evaluate")

def calculate(params: CalculateParams) -> float:
    """Evaluate a mathematical expression."""
    return eval(params.expression)

# Create server
server = MCPServer(
    name="assistant-tools",
    version="1.0.0",
    description="Tools for AI assistants"
)

# Add tools
server.add_tools(mcp_from_functions([
    get_weather,
    search_web,
    calculate,
]))

```

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

### Pattern 3: Environment Setup with Shell Middleware

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/guides/shell-tool-guide.md

Automate the setup of development environments using ShellMiddleware. This pattern can handle tasks such as checking Python versions, creating virtual environments, installing dependencies, and setting up pre-commit hooks.

```python
from ai_infra import Agent
from ai_infra.llm.shell import ShellMiddleware

agent = Agent(
    deep=True,
    middleware=[ShellMiddleware(workspace_root="/project")]
)

# Automated development environment setup
result = agent.run("""
    Set up the development environment:
    1. Check Python version
    2. Create virtual environment if not exists
    3. Install dependencies from pyproject.toml
    4. Set up pre-commit hooks
    5. Verify the setup by running a simple test
""")
```

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

### Quick Start Logging

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/infrastructure/logging.md

Initialize the logging system and log a message with structured context.

```python
from ai_infra import configure_logging, get_logger

configure_logging(level="INFO")
logger = get_logger(__name__)

logger.info("Processing request", request_id="abc123", user_id="user1")
```

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

### Quick Start CopilotAgent

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/copilot-agent.md

Initialize CopilotAgent and run a task. Ensure the 'copilot' extra is installed and GitHub Copilot CLI is authenticated.

```python
from ai_infra import CopilotAgent

agent = CopilotAgent(cwd="/path/to/project")
result = await agent.run("Add type hints to all public functions in src/")
print(result.content)
```

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

### Run Async Examples

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Execute asynchronous Python example scripts using the asyncio module.

```bash
python chat/02_streaming.py
```

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

### Install ai-infra

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/getting-started.md

Install the library using pip or Poetry.

```bash
pip install ai-infra
```

```bash
poetry add ai-infra
```

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

### Initialize RealtimeVoice

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/realtime.md

Examples for initializing the RealtimeVoice client with auto-detection or explicit provider selection.

```python
from ai_infra import RealtimeVoice

voice = RealtimeVoice()  # Auto-detects provider

async with voice.connect() as session:
    # Send audio, get audio back
    response = await session.send_audio(audio_bytes)
    play_audio(response.audio)
```

```python
from ai_infra import RealtimeVoice

# Auto-detects from env vars (OpenAI -> Google)
voice = RealtimeVoice()
```

```python
# OpenAI Realtime
voice = RealtimeVoice(provider="openai")

# Google Gemini Live
voice = RealtimeVoice(provider="gemini")
```

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

### MCP Client Configuration Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/src/ai_infra/cli/README.md

An example JSON configuration for an MCP client, defining how to connect to a server using npx with specified arguments and environment variables.

```json
{
  "servers": {
    "<FriendlyName>": {
      "command": "npx",
      "args": ["--yes","--package=github:<OWNER>/<REPO>","<TOOL_NAME>"],
      "env": { "UVX_PATH": "/abs/path/to/uvx" }
    }
  }
}
```

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

### Error Handling Quick Start

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/infrastructure/errors.md

A quick example demonstrating how to catch common AI infrastructure errors like RateLimitError, ContentFilterError, and general LLMError.

```APIDOC
## Error Handling Quick Start

### Description
This section provides a quick start guide to handling common errors that may occur during AI operations using the `ai-infra` library. It demonstrates how to use `try-except` blocks to catch specific exceptions like `RateLimitError`, `ContentFilterError`, and the base `LLMError`.

### Method
N/A (Illustrative Code)

### Endpoint
N/A (Illustrative Code)

### Parameters
N/A

### Request Example
```python
from ai_infra import LLMError, RateLimitError, ContentFilterError

try:
    response = await llm.generate(prompt)
except RateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except ContentFilterError as e:
    print(f"Content blocked: {e.reason}")
except LLMError as e:
    print(f"LLM error: {e}")
```

### Response
N/A (Illustrative Code)
```

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

### Initialize and Run an MCP Server

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/server.md

Basic setup for adding resources and running an MCP server.

```python
server.add_resource(
    uri="info://about",
    name="About",
    mime_type="text/plain",
    content="This server provides weather, search, and calculation tools."
)

# Run
if __name__ == "__main__":
    server.run(port=8080)
```

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

### Quick Start with LLM

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/core/llm.md

Initialize the LLM class with zero configuration to auto-detect providers from environment variables.

```python
from ai_infra import LLM

# Zero-config: auto-detects from environment
llm = LLM()
response = llm.chat("What is 2+2?")
print(response)  # "4"
```

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

### Install CopilotAgent

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/copilot-agent.md

Install the 'ai-infra' package with the 'copilot' extra. Also, install and authenticate the GitHub Copilot CLI.

```bash
pip install 'ai-infra[copilot]'
```

```bash
# macOS / Linux via npm
npm install -g @github/copilot-cli

# Authenticate
gh auth login
gh copilot --version
```

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

### Quick Start Workspace Usage

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/workspace.md

Initialize a workspace and perform basic file operations.

```python
from ai_infra import Workspace

workspace = Workspace("./project")

# Read files
content = workspace.read("src/main.py")

# Write files
workspace.write("output/result.txt", "Hello World")

# Search files
matches = workspace.search("TODO")
```

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

### Install Document Support Extras

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/embeddings/retriever.md

Install the necessary extras for supporting various file types like PDF, Word, and CSV.

```bash
pip install ai-infra[documents]
```

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

### Configure Entry and Compilation

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/core/graph.md

Set the starting node and compile the graph for execution.

```python
# Set where the graph starts
graph.set_entry_point("fetch_context")

# Compile the graph
compiled = graph.compile()
```

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

### Install ai-infra CLI

Source: https://github.com/nfraxlab/ai-infra/blob/main/src/ai_infra/cli/README.md

Installs the ai-infra CLI using pip or Poetry. Poetry installation includes a help command.

```bash
pip install ai-infra
```

```bash
poetry install
poetry run ai-infra --help
```

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

### Quick Start with tools_from_object

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/object-tools.md

Demonstrates basic usage of converting a Calculator class instance into tools for an Agent.

```python
from ai_infra import Agent, tools_from_object

class Calculator:
    """A simple calculator."""

    def add(self, a: float, b: float) -> float:
        """Add two numbers."""
        return a + b

    def multiply(self, a: float, b: float) -> float:
        """Multiply two numbers."""
        return a * b

calc = Calculator()
tools = tools_from_object(calc)

agent = Agent(tools=tools)
result = agent.run("What is 5 + 3?")
```

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

### Install ai-infra and Set API Keys

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Install the ai-infra library in development mode and set necessary API keys for different AI providers. Alternatively, API keys can be set in a .env file.

```bash
cd examples
pip install -e ..  # Install ai-infra in development mode
# or: poetry install

# Set your API keys
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."
export GOOGLE_API_KEY="..."
```

```bash
# examples/.env
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=...
```

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

### Install ai-infra package

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/troubleshooting.md

Ensure the library is installed in your environment using pip or poetry.

```bash
# Make sure you installed it
pip install ai-infra

# Or in the correct virtual environment
poetry add ai-infra
```

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

### Configure OpenAI API and Start Server

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/openapi.md

Registers an OpenAPI specification from a remote URL and starts the server on a specified port.

```python
server.add_openapi(
    "https://raw.githubusercontent.com/openai/openai-openapi/master/openapi.yaml",
    auth={"Authorization": "Bearer sk-xxx"},
    tags=["Chat"],
    prefix="openai_"
)

# Run
server.run(port=8080)
```

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

### Basic LLM Chat Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

A fundamental example demonstrating how to initialize an LLM and send a simple chat message. Requires an API key for the LLM provider to be set in the environment.

```python
#!/usr/bin/env python
"""Example title and description.

This example demonstrates:
- Feature 1
- Feature 2
- Feature 3

Required API Keys:
- OPENAI_API_KEY (or your preferred provider)
"""

from ai_infra import LLM

def main():
    # Example code here
    llm = LLM()
    response = llm.chat("Hello!")
    print(response.content)

if __name__ == "__main__":
    main()

```

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

### Quick Start VectorStore

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/embeddings/vectorstore.md

Initialize a VectorStore, add documents, and perform a basic similarity search.

```python
from ai_infra import VectorStore

store = VectorStore()
store.add("doc1", "Python is a programming language")
store.add("doc2", "JavaScript runs in browsers")

results = store.search("coding languages")
print(results[0].content)  # "Python is a programming language"
```

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

### Quick Start Progress Streaming

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/progress.md

Demonstrates defining a tool with the @progress decorator and registering it with an Agent.

```python
from ai_infra import Agent, progress

@progress
def process_files(files: list[str]) -> str:
    """Process multiple files with progress updates."""
    for i, file in enumerate(files):
        yield {"progress": (i + 1) / len(files), "status": f"Processing {file}"}
        do_processing(file)
    return "All files processed"

agent = Agent(tools=[process_files])
```

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

### Install and Run Benchmark Suite

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/performance.md

Install pytest-benchmark to run the ai-infra benchmark suite. Use the --benchmark-only flag to execute the benchmarks and --benchmark-compare to compare results against previous runs.

```bash
# Install benchmark dependencies
pip install pytest-benchmark

# Run benchmark suite
pytest benchmarks/ --benchmark-only

# Compare to baseline
pytest benchmarks/ --benchmark-compare
```

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

### Provider-Specific Vision Configuration

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/vision.md

Examples for initializing specific providers like OpenAI, Anthropic, and Google.

```python
llm = LLM(provider="openai", model="gpt-4o")

response = llm.chat(
    "What's in this image?",
    images=["photo.jpg"]
)
```

```python
llm = LLM(provider="anthropic", model="claude-sonnet-4-20250514")

response = llm.chat(
    "Analyze this image",
    images=["photo.jpg"]
)
```

```python
llm = LLM(provider="google_genai", model="gemini-2.0-flash")

response = llm.chat(
    "Describe what you see",
    images=["photo.jpg"]
)
```

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

### Quick Start Image Generation

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/imagegen/imagegen.md

Initialize ImageGen and generate a simple image. The provider is auto-detected from environment variables.

```python
from ai_infra import ImageGen

gen = ImageGen()  # Auto-detects provider from env
image = gen.generate("A serene mountain landscape at sunset")
image.save("landscape.png")
```

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

### Implement on_agent_start Lifecycle Hook

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-middleware.md

Example implementation of the on_agent_start hook to initialize a shell session.

```python
async def on_agent_start(self, agent: Agent) -> None:
    """Initialize shell session when agent starts."""
    self._session = ShellSession(
        workspace_root=self.workspace_root,
        timeout=self.session_timeout,
    )
    await self._session.start()
```

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

### Configure API Keys

Source: https://github.com/nfraxlab/ai-infra/blob/main/README.md

Environment variable setup for various AI providers.

```bash
# Set your API keys (use whichever providers you need)
export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...
export GOOGLE_API_KEY=...
```

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

### Using Built-in Callbacks

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/callbacks.md

Examples of initializing and using the provided Logging, Metrics, and Print callback implementations.

```python
from ai_infra import LoggingCallbacks, Agent

agent = Agent(tools=[...], callbacks=LoggingCallbacks())
```

```python
from ai_infra import MetricsCallbacks

metrics = MetricsCallbacks()
agent = Agent(tools=[...], callbacks=metrics)

# After running...
print(f"Total calls: {metrics.total_calls}")
print(f"Total tokens: {metrics.total_tokens}")
```

```python
from ai_infra import PrintCallbacks

agent = Agent(tools=[...], callbacks=PrintCallbacks())
```

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

### Basic Chat Completion Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

A simple example demonstrating basic chat completion with an LLM. Ensure API keys are set.

```python
from ai_infra.chat import Chat

chat = Chat()
response = chat.complete("Hello, how are you?")
print(response)
```

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

### Documentation Agent Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/deep-agent.md

Set up a DeepAgent to generate documentation. The agent's goal includes analyzing modules, extracting docstrings, generating API references, and creating usage examples. It utilizes tools for file operations and directory listing.

```python
agent = DeepAgent(
    goal="""
    Generate comprehensive documentation:
    1. Analyze all modules and functions
    2. Extract docstrings and type hints
    3. Generate API reference
    4. Create usage examples
    5. Write getting started guide
    """,
    workspace=workspace,
    tools=[read_file, write_file, search, list_dir],
)

result = await agent.run()
```

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

### Usage Patterns

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-middleware.md

Examples of integrating ShellMiddleware with Agents.

```APIDOC
## Usage Patterns

### With Agent

```python
from ai_infra import Agent
from ai_infra.llm.shell import ShellMiddleware

middleware = ShellMiddleware(
    workspace_root="/path/to/project"
)

agent = Agent(
    deep=True,
    middleware=[middleware]
)

# Agent has access to persistent shell
result = agent.run("""
    1. Create a virtual environment
    2. Install dependencies from requirements.txt
    3. Run pytest
""")
```

### With Agent and Customization

```python
from ai_infra import Agent
from ai_infra.llm.shell import ShellMiddleware

# Provide custom middleware
middleware = ShellMiddleware(
    workspace_root="/project",
    allowed_commands=("pytest", "make"),
)

agent = Agent(
    deep=True,
    middleware=[middleware],
)
```

### Multiple Middlewares

```python
from ai_infra import Agent
from ai_infra.llm.shell import ShellMiddleware
from ai_infra.llm.middleware import LoggingMiddleware

agent = Agent(
    deep=True,
    middleware=[
        LoggingMiddleware(),
        ShellMiddleware(workspace_root="/project"),
    ]
)
```
```

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

### Initialize Shell Tool in Agent

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-tool.md

Basic setup to integrate the run_shell tool into an AI agent.

```python
from ai_infra import Agent
from ai_infra.llm.shell import run_shell

agent = Agent(tools=[run_shell])
result = agent.run("List all Python files in the current directory")
```

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

### Async Usage

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/tts.md

An example demonstrating how to use the TTS client asynchronously.

```APIDOC
## Async Usage

```python
import asyncio
from ai_infra import TTS

async def main():
    tts = TTS()
    audio = await tts.aspeak("Async speech!")
    audio.save("output.mp3")

asyncio.run(main())
```
```

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

### Structured Output with Pydantic

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Example of obtaining structured output (JSON/Pydantic) from a chat model. Requires Pydantic to be installed.

```python
from pydantic import BaseModel
from ai_infra.chat import Chat

class Person(BaseModel):
    name: str
    age: int

chat = Chat()
person = chat.complete_with_schema(Person, "Create a person named Alice who is 30 years old.")
print(person)
```

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

### Quick Start with Pydantic Models

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/schema-tools.md

Initialize an agent with CRUD tools generated from a Pydantic model.

```python
from pydantic import BaseModel
from ai_infra import Agent, tools_from_models

class User(BaseModel):
    id: int
    name: str
    email: str

tools = tools_from_models([User])
agent = Agent(tools=tools)

result = agent.run("Create a user named John with email john@example.com")
```

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

### Generate Batch Product Images

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/imagegen/imagegen.md

Example of iterating through styles to generate multiple product images with specific provider configurations.

```python
async def generate_product_images(
    product_name: str,
    styles: list[str],
) -> list:
    results = []

    for style in styles:
        result = await generate_image(
            prompt=f"Professional product photo of {product_name}, {style} style, white background, studio lighting",
            provider="openai",
            model_name="dall-e-3",
            size="1024x1024",
            quality="hd",
        )
        results.append(result)

    return results

# Generate in multiple styles
images = await generate_product_images(
    "wireless headphones",
    styles=["minimalist", "lifestyle", "technical"],
)
```

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

### Configure ShellMiddleware Options

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-middleware.md

Various configuration examples for ShellMiddleware, including workspace root, timeouts, redaction, and security settings.

```python
middleware = ShellMiddleware(
    workspace_root="/home/user/projects/my-app"
)
# All commands start from /home/user/projects/my-app
```

```python
middleware = ShellMiddleware(
    workspace_root="/project",
    session_timeout=7200.0  # 2 hours
)
```

```python
from ai_infra.llm.shell import RedactionRule

custom_rules = (
    RedactionRule(
        name="my_api_key",
        pattern=r"MY_KEY_[A-Za-z0-9]{32}",
        replacement="[REDACTED:my_key]"
    ),
    RedactionRule(
        name="internal_token",
        pattern=r"internal-token-[a-f0-9]+",
        replacement="[REDACTED:internal_token]"
    ),
)

middleware = ShellMiddleware(
    workspace_root="/project",
    redaction_rules=custom_rules,
)
```

```python
# Disable dangerous pattern checking (use with caution)
middleware = ShellMiddleware(
    workspace_root="/project",
    dangerous_pattern_check=False,
)
```

```python
import re

custom_patterns = (
    re.compile(r"docker\s+rm\s+-f"),  # Block force-remove containers
    re.compile(r"kubectl\s+delete"),   # Block k8s deletions
)

middleware = ShellMiddleware(
    workspace_root="/project",
    custom_dangerous_patterns=custom_patterns,
)
```

```python
middleware = ShellMiddleware(
    workspace_root="/project",
    allowed_commands=("pytest", "npm", "poetry", "make"),
)
# Only commands starting with these prefixes are allowed
```

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

### STT Word Timestamps

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/stt.md

Enable word-level timestamps during transcription and iterate through the results to get the text, start, and end times for each word.

```python
result = stt.transcribe("audio.mp3", timestamps=True)

for word in result.words:
    print(f"{word.text}: {word.start}s - {word.end}s")
```

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

### Initialize an Agent with Tools

Source: https://github.com/nfraxlab/ai-infra/blob/main/README.md

Demonstrates creating an agent with custom tools that can be executed across different LLM providers.

```python
from ai_infra import Agent

def search_web(query: str) -> str:
    """Search the web."""
    return f"Results for: {query}"

agent = Agent(tools=[search_web])
result = agent.run("Find the latest news about AI")
# Works with OpenAI, Anthropic, Google—same code.
```

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

### Execute Shell Commands with Various Syntax

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-tool.md

Examples of executing commands using pipes, chaining, and redirects.

```python
# Simple command
result = await run_shell.ainvoke({"command": "ls -la"})

# Pipes
result = await run_shell.ainvoke({"command": "cat file.txt | grep 'error'"})

# Command chaining
result = await run_shell.ainvoke({"command": "cd /tmp && ls"})

# Redirects
result = await run_shell.ainvoke({"command": "echo 'hello' > output.txt"})
```

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

### Quick Start Retriever Initialization and Usage

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/embeddings/retriever.md

Initialize the Retriever, add documents, and perform a search. This is the basic entry point for using the Retriever.

```python
from ai_infra import Retriever

retriever = Retriever()

# Add documents
retriever.add("Python is a programming language")
retriever.add("JavaScript runs in browsers")

# Search
results = retriever.search("coding")
print(results[0].content)  # "Python is a programming language"
```

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

### List Available Voices

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/realtime.md

Call `list_voices` to get a list of available voice identifiers for a given provider. Example shows listing voices for 'openai'.

```python
voices = RealtimeVoice.list_voices("openai")
# ['alloy', 'ash', 'ballad', ...]
```

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

### RAG with Pinecone Managed Cloud

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Example of configuring the retriever to use Pinecone, a managed cloud vector database service. Requires Pinecone API key and environment setup.

```python
from ai_infra.retriever import Retriever

retriever = Retriever(backend="pinecone", index_name="my-index")
retriever.add("Data stored in Pinecone.")

results = retriever.search("query for Pinecone data")
print(results)
```

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

### Create Simple MCP Server

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Example of setting up a basic Model Context Protocol (MCP) server. This server can expose tools to clients.

```python
from ai_infra.mcp.server import MCPServer

def add(a: int, b: int) -> int:
    return a + b

server = MCPServer(tools=[add])
server.run(port=8000) # Run the server on port 8000
```

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

### Use Cases for Callbacks

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/callbacks.md

Code examples demonstrating how to implement custom callbacks for various use cases.

```APIDOC
## Use Cases

### Observability & Monitoring

```python
class ObservabilityCallbacks(Callbacks):
    def on_llm_start(self, event):
        span = tracer.start_span("llm_call")
        span.set_attribute("provider", event.provider)
        span.set_attribute("model", event.model)

    def on_llm_end(self, event):
        span.set_attribute("tokens", event.total_tokens)
        span.set_attribute("latency_ms", event.latency_ms)
        span.end()

    def on_llm_error(self, event):
        span.record_exception(event.error)
        span.end()
```

### Cost Tracking

```python
class CostCallbacks(Callbacks):
    COSTS = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "claude-3-opus": {"input": 0.015, "output": 0.075},
    }

    def __init__(self):
        self.total_cost = 0.0

    def on_llm_end(self, event):
        if event.model in self.COSTS:
            cost = self.COSTS[event.model]
            self.total_cost += (
                (event.input_tokens or 0) * cost["input"] / 1000 +
                (event.output_tokens or 0) * cost["output"] / 1000
            )
```

### Progress Reporting

```python
class ProgressCallbacks(Callbacks):
    async def on_mcp_progress_async(self, event):
        if event.total:
            percentage = event.progress / event.total * 100
            print(f"[{event.server_name}] {percentage:.0f}% - {event.message}")
```
```

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

### Quick Start Vision Chat

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/vision.md

Initialize the LLM client and perform a basic chat request with an image.

```python
from ai_infra import LLM

llm = LLM()
response = llm.chat(
    "What's in this image?",
    images=["photo.jpg"]
)
print(response)
```

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

### Quick Start: Record, Save, and Replay Agent Workflow

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/replay.md

This snippet demonstrates the complete cycle of recording an agent's workflow, saving the trace to a file, and then loading and iterating through the saved trace for analysis.

```python
from ai_infra import Agent, Replay

# 1. Run agent with recording enabled
agent = Agent(tools=[search, write_file], record=True)
result = agent.run("Search for Python tutorials and save notes")

# 2. Get the recorded trace
trace = agent.get_trace()
trace.save("my_workflow.json")

# 3. Later, replay and analyze
replay = Replay.from_file("my_workflow.json")
for step in replay.steps():
    print(f"Step {step.number}: {step.action}")
```

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

### MCPServer Initialization with Description

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/server.md

Initialize an MCPServer with a name, version, and a descriptive string. This helps clients understand the server's purpose.

```python
server = MCPServer(
    name="weather-server",
    version="1.0.0",
    description="Provides weather information"
)
```

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

### Install ai-infra CLI

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/cli.md

Install the ai-infra CLI using pip. This command is typically run once.

```bash
pip install ai-infra
```

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

### Basic MCPServer Initialization

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/server.md

Create a basic MCPServer instance. This is the foundation for exposing tools and resources.

```python
from ai_infra import MCPServer

server = MCPServer(
    name="my-server",
    version="1.0.0"
)
```

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

### Conventional commit examples

Source: https://github.com/nfraxlab/ai-infra/blob/main/CONTRIBUTING.md

Examples of correctly formatted commit messages following the Conventional Commits specification.

```text
feat: add streaming support for agents
fix: handle timeout in MCP client
docs: update getting-started guide
refactor: extract callback normalization to shared utility
test: add unit tests for memory module
```

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

### Invalid commit message examples

Source: https://github.com/nfraxlab/ai-infra/blob/main/CONTRIBUTING.md

Examples of commit messages that lack the required type prefix and will be grouped as 'Other Changes'.

```text
Refactor code for improved readability  <- Missing type prefix!
updating docs                           <- Missing type prefix!
bug fix                                 <- Missing type prefix!
```

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

### Quick Start: Initialize Deep Agent

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/deep-agent.md

Initialize a Deep Agent with a goal and a list of tools for autonomous task execution.

```python
from ai_infra import DeepAgent

agent = DeepAgent(
    goal="Build a complete REST API for user management",
    tools=[search, read_file, write_file, run_tests],
)

result = await agent.run()
```

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

### Get Structured Output from LLM

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/quickstart.md

Use Pydantic models to get structured output from the LLM. The response is parsed into the specified model.

```python
from pydantic import BaseModel
from ai_infra import LLM

class Answer(BaseModel):
    value: int
    explanation: str

llm = LLM()
result = llm.chat("What is 2 + 2?", response_model=Answer)
print(result.value)        # 4
print(result.explanation)  # "Two plus two equals four"
```

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

### Test Specific Category with Pytest

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Run tests for the examples directory using pytest, useful for verifying example functionality if test mode is enabled.

```bash
python -m pytest examples/ -v
```

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

### Vision/Multimodal Chat Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/examples/README.md

Example of using vision capabilities for multimodal chat, allowing interaction with images. Ensure your provider supports vision models.

```python
from ai_infra.chat import Chat

chat = Chat()
response = chat.complete_vision("What is in this image?", image_url="https://example.com/image.jpg")
print(response)
```

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

### Quick Start: Create and Run Agent with Persona

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/personas.md

Load a persona from a YAML file and initialize an agent with it. Then, run the agent with a specific task.

```python
from ai_infra import Agent, Persona

persona = Persona.from_yaml("analyst.yaml")
agent = Agent(persona=persona)
result = agent.run("Analyze this data")
```

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

### Handle Shell Session Start Error

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-middleware.md

Demonstrates error handling for cases where the shell session fails to start, catching a `ShellSessionError` and printing a user-friendly message.

```python
# If session fails to start
try:
    await middleware.on_agent_start(agent)
except ShellSessionError as e:
    print(f"Failed to start shell session: {e}")
```

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

### Implement Production Agent Deployment

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/memory.md

Full implementation example including session storage, memory stores, and conversation indexing for production environments.

```python
import os
from ai_infra import Agent
from ai_infra.llm.session import postgres as session_postgres
from ai_infra.llm.memory import MemoryStore, SummarizationMiddleware
from ai_infra.llm.tools.custom import ConversationMemory, create_memory_tool

DATABASE_URL = os.environ["DATABASE_URL"]

# 1. Session storage (short-term, per-thread)
session_storage = session_postgres(DATABASE_URL)

# 2. Memory store (long-term facts)
memory_store = MemoryStore.postgres(DATABASE_URL)

# 3. Conversation memory (long-term conversation history)
conv_memory = ConversationMemory(
    backend="postgres",
    connection_string=DATABASE_URL,
    embedding_provider="openai",
    embedding_model="text-embedding-3-small",
)

# 4. Create recall tool
def get_recall_tool(user_id: str):
    return create_memory_tool(
        conv_memory,
        user_id=user_id,
        limit=5,
    )

# 5. Create agent
def create_agent_for_user(user_id: str) -> Agent:
    return Agent(
        tools=[get_recall_tool(user_id)],
        session=session_storage,
        system="You are a helpful assistant. Use recall_past_conversations when users reference previous discussions.",
    )

# Usage
agent = create_agent_for_user("user_123")

# Run with session (conversation persists)
result = agent.run(
    "How did we fix that bug last week?",
    session_id="thread_789",
)

# After conversation ends, index it for future recall
# (Call this when session ends or periodically)
conv_memory.index_conversation(
    user_id="user_123",
    session_id="thread_789",
    messages=result.messages,  # Get from session
    metadata={"date": "2025-01-20"},
)
```

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

### Analyze Errors and Get Context

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/replay.md

Check if a replay encountered an error, retrieve details about the error step, and get the context (preceding steps) before the error occurred.

```python
if replay.has_error:
    error_step = replay.error_step
    print(f"Error at step {error_step.number}")
    print(f"Error: {error_step.error}")
    print(f"Traceback: {error_step.traceback}")

    # Get context before error
    context = replay.get_context_before(error_step.number, steps=3)
```

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

### FastAPI WebSocket Example for Streaming Agent Responses

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/streaming.md

A FastAPI WebSocket example to stream agent responses. It accepts a message, streams events, and sends them as JSON over the WebSocket.

```python
from fastapi import WebSocket

@app.websocket("/chat")
def chat_ws(websocket: WebSocket):
    await websocket.accept()
    message = await websocket.receive_text()

    async for event in agent.astream(message):
        await websocket.send_json(event.to_dict())

    await websocket.close()
```

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

### MemoryStore Backends

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/memory.md

Illustrates initializing `MemoryStore` with different backends: in-memory for development, SQLite for single-instance production, and PostgreSQL for multi-instance production.

```python
from ai_infra.llm.memory import MemoryStore

# In-memory (development, testing)
store = MemoryStore()

# SQLite (single-instance production)
store = MemoryStore.sqlite("./memories.db")

# PostgreSQL (multi-instance production)
store = MemoryStore.postgres(os.environ["DATABASE_URL"])
```

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

### Quick Start MCP Client Connection

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/mcp/client.md

Connect to an MCP server using HTTP and perform basic tool discovery and execution.

```python
from ai_infra import MCPClient

async with MCPClient("http://localhost:8080") as client:
    tools = await client.list_tools()
    result = await client.call_tool("get_weather", {"city": "NYC"})
```

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

### RealtimeVoice Initialization

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/multimodal/realtime.md

Demonstrates how to initialize the RealtimeVoice client, either by auto-detecting the provider or by explicitly specifying one.

```APIDOC
## RealtimeVoice Initialization

### Description
Initialize the RealtimeVoice client. It can auto-detect the provider based on environment variables or be configured with a specific provider.

### Method
`RealtimeVoice()`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from ai_infra import RealtimeVoice

# Auto-detect provider (e.g., from environment variables)
voice_auto = RealtimeVoice()

# Explicitly specify provider
voice_openai = RealtimeVoice(provider="openai")
voice_google = RealtimeVoice(provider="gemini")
```

### Response
None (initialization)

### Error Handling
- If no provider is configured or detected, an error may occur.
- Invalid provider names will raise an error.
```

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

### Emit Turn Start Event

Source: https://github.com/nfraxlab/ai-infra/blob/main/ROADMAP.md

Emit a 'turn_start' StreamEvent at the beginning of each agent iteration to signal the start of a new reasoning or tool-calling cycle. The event includes a turn ID.

```python
_turn_id = 0
# At start of each iteration in the tool-calling loop:
_turn_id += 1
if should_emit_event("turn_start", eff_visibility):
    yield StreamEvent(type="turn_start", turn_id=_turn_id)
```

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

### FastAPI SSE Example for Streaming Agent Responses

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/streaming.md

A complete FastAPI Server-Sent Events (SSE) example to stream agent responses. It yields JSON-encoded events for real-time UI updates.

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from ai_infra import Agent
import json

app = FastAPI()
agent = Agent(tools=[search_docs])

@app.post("/chat")
async def chat(message: str):
    async def generate():
        async for event in agent.astream(message, visibility="detailed"):
            yield f"data: {json.dumps(event.to_dict())}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")
```

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

### List and Get Providers

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/core/providers.md

Demonstrates how to list all registered providers, find providers for a specific capability, and retrieve detailed configuration for a given provider.

```python
from ai_infra import (
    ProviderRegistry,
    ProviderCapability,
    list_providers,
    list_providers_for_capability,
    get_provider,
    is_provider_configured,
)

# List all registered providers
providers = list_providers()
# ['openai', 'anthropic', 'google_genai', 'xai', 'elevenlabs', ...]

# List providers for a specific capability
chat_providers = list_providers_for_capability(ProviderCapability.CHAT)
# ['openai', 'anthropic', 'google_genai', 'xai']

# Get provider configuration
openai = get_provider("openai")
print(openai.display_name)  # "OpenAI"
print(openai.env_var)       # "OPENAI_API_KEY"

# Check if provider is configured (API key set)
if is_provider_configured("openai"):
    print("OpenAI is ready!")
```

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

### ShellMiddleware Initialization and Usage

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/tools/shell-middleware.md

Demonstrates how to initialize ShellMiddleware with a workspace root and use it with an AI agent.

```APIDOC
## POST /api/users

### Description
This endpoint allows for the creation of a new user in the system.

### Method
POST

### Endpoint
/api/users

### Parameters
#### Query Parameters
- **name** (string) - Required - The name of the user.
- **email** (string) - Required - The email address of the user.

### Request Body
```json
{
  "username": "john_doe",
  "password": "securepassword123"
}
```

### Response
#### Success Response (201)
- **id** (integer) - The unique identifier for the newly created user.
- **message** (string) - A confirmation message.

#### Response Example
```json
{
  "id": 123,
  "message": "User created successfully."
}
```
```

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

### Configure Tracing with Default Settings

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/infrastructure/tracing.md

Quickly set up tracing for your AI service. This example configures the service name and the OTLP endpoint for sending trace data.

```python
from ai_infra import configure_tracing, LLM

configure_tracing(
    service_name="my-ai-service",
    endpoint="http://jaeger:4317",
)

llm = LLM(provider="openai")
response = await llm.generate("Hello")  # Automatically traced
```

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

### Consumer Usage of Published MCP Shim

Source: https://github.com/nfraxlab/ai-infra/blob/main/src/ai_infra/cli/README.md

Demonstrates how consumers can run a published MCP stdio server using npx, including specifying the repository and tool name. Also shows how to set UVX_PATH if uvx is not on the system PATH.

```bash
npx --y --package=github:<OWNER>/<REPO> <TOOL_NAME> [args...]
```

```bash
UVX_PATH=/abs/path/to/uvx npx --yes --package=github:<OWNER>/<REPO> <TOOL_NAME>
```

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

### GET /models

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/copilot-agent.md

List all available models for the agent.

```APIDOC
## GET /models

### Description
Retrieve a list of available models supported by the agent.

### Method
GET

### Endpoint
agent.list_models()

### Response
#### Success Response (200)
- **models** (list[str]) - A list of model names.
```

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

### Code Review Agent Example

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/deep-agent.md

Configure a DeepAgent to perform code reviews. This example specifies a goal for reviewing Python files, identifies potential issues, suggests improvements, and generates a report. It also sets a specific provider, model, and max iterations.

```python
from ai_infra import DeepAgent, Workspace

workspace = Workspace("./project")

agent = DeepAgent(
    goal="""
    Review all Python files in the codebase:
    1. Check for code quality issues
    2. Identify potential bugs
    3. Suggest improvements
    4. Generate a report in docs/code-review.md
    """,
    workspace=workspace,
    tools=[read_file, write_file, search, run_linter],
    provider="anthropic",
    model_name="claude-sonnet-4-20250514",
    max_iterations=100,
)

result = await agent.run()
print(result.output)
```

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

### Complete DeepAgent Example with Features

Source: https://github.com/nfraxlab/ai-infra/blob/main/docs/features/deep-agent.md

Demonstrates all DeepAgent features, including custom tool definitions, workspace integration, planning parameters like `max_iterations` and `allow_self_correction`, and human approval for dangerous operations. It also shows how to stream progress and handle approval requests.

```python
from ai_infra import DeepAgent, Workspace

# 1. Set up workspace with sandboxed file access
workspace = Workspace("./my-project")

# 2. Define tools the agent can use
def search_codebase(query: str) -> str:
    """Search for code patterns."""
    return workspace.search(query)

def read_file(path: str) -> str:
    """Read a file."""
    return workspace.read(path)

def write_file(path: str, content: str) -> str:
    """Write to a file."""
    workspace.write(path, content)
    return f"Wrote {len(content)} bytes to {path}"

def run_tests() -> str:
    """Run the test suite."""
    # This is a dangerous operation - will require approval
    import subprocess
    result = subprocess.run(["pytest", "-q"], capture_output=True, text=True)
    return result.stdout

# 3. Create agent with all features
agent = DeepAgent(
    goal="Add type hints to all functions in src/ and verify with tests",
    tools=[search_codebase, read_file, write_file, run_tests],
    workspace=workspace,

    # Planning & Execution
    max_iterations=50,
    allow_self_correction=True,  # Retry on failures

    # Human approval for dangerous operations
    require_approval_for=["run_tests", "write_file"],
)

# 4. Run with progress tracking
async def main():
    async for progress in agent.stream():
        # Show progress
        print(f"[Step {progress.step}] {progress.status}")

        # Handle approval requests
        if progress.requires_approval:
            print(f"\n[!]  Approval needed: {progress.pending_action}")
            print(f"   Tool: {progress.tool_name}")
            print(f"   Args: {progress.tool_args}")

            if input("Approve? [y/n]: ").lower() == "y":
                await agent.approve()
            else:
                await agent.reject("User declined")

    # Get final result
    result = agent.get_result()

    # Summary
    print(f"\n[OK] Completed in {result.steps_count} steps")
    print(f"   Duration: {result.duration_seconds:.1f}s")
    print(f"   Tokens: {result.total_tokens}")
    print(f"   Cost: ${result.estimated_cost:.4f}")
    print(f"\nOutput:\n{result.output}")

import asyncio
asyncio.run(main())
```