# Multimodal Content
Source: https://moxn.mintlify.app/advanced/multimodal
Work with images and documents in prompts
Moxn supports multimodal prompts with images, PDFs, and other files. This guide covers how to include and handle multimodal content.
## Supported Content Types
| Type | Providers | Use Cases |
| ---------- | ----------------- | ----------------------------- |
| **Images** | All | Charts, screenshots, diagrams |
| **PDFs** | Anthropic, Google | Documents, reports |
| **Files** | Varies | Various document types |
## Images in Messages
### In the Web App
Add images to messages:
1. Click the image icon in the editor
2. Upload an image or paste a URL
3. Add alt text for accessibility
### In Code
Images appear as content blocks:
```python theme={null}
# When you fetch a prompt with images
prompt = await client.get_prompt("...", branch_name="main")
for message in prompt.messages:
for block_group in message.blocks:
for block in block_group:
if block.block_type == "image_from_source":
print(f"Image: {block.url}")
print(f"Alt: {block.alt}")
```
### Provider Conversion
Images are automatically converted to provider format:
```python theme={null}
# Anthropic format (base64)
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": "iVBORw0KGgo..."
}
}
# OpenAI format (data URI)
{
"type": "image_url",
"image_url": {
"url": "data:image/png;base64,iVBORw0KGgo..."
}
}
# Google format (Part with blob)
Part(inline_data=Blob(
mime_type="image/png",
data=bytes(...)
))
```
## PDFs in Messages
### Adding PDFs
In the web app:
1. Click the file icon
2. Upload a PDF
3. It appears inline in the message
### PDF Block Types
Like images, PDFs use `PDFContentFromSource`—a discriminated union. You can pass a dict or explicit type:
```python theme={null}
pdf = {
"type": "url",
"url": "https://example.com/document.pdf",
"media_type": "application/pdf",
"filename": "document.pdf"
}
```
```python theme={null}
from moxn.base_models.blocks.file import MediaDataPDFFromURL
pdf = MediaDataPDFFromURL(
url="https://example.com/document.pdf",
media_type="application/pdf",
filename="document.pdf"
)
```
### PDF Types
| `type` value | Class | Required Fields |
| -------------- | --------------------------- | ------------------------------------ |
| `"url"` | `MediaDataPDFFromURL` | `url`, `media_type`, `filename` |
| `"base64"` | `MediaDataPDFFromBase64` | `base64`, `media_type`, `filename` |
| `"bytes"` | `MediaDataPDFFromBytes` | `bytes`, `media_type`, `filename` |
| `"local_file"` | `MediaDataPDFFromLocalFile` | `filepath`, `media_type`, `filename` |
## Signed URLs
For content stored in cloud storage (S3, GCS, etc.), Moxn uses signed URLs with automatic refresh.
### How It Works
```
1. You upload content → stored in cloud storage
2. Prompt fetched → signed URLs generated (short expiry)
3. SDK registers content → tracks expiration
4. Before expiry → auto-refreshes URLs
5. Provider conversion → fresh URLs used
```
### Automatic Refresh
The SDK handles refresh automatically:
```python theme={null}
async with MoxnClient() as client:
# Signed URLs are registered when fetching
prompt = await client.get_prompt("...", branch_name="main")
# Later, when converting (URLs refreshed if needed)
session = PromptSession.from_prompt_template(prompt, session_data)
payload = session.to_anthropic_invocation()
# ^ URLs are fresh at this point
```
### Manual Registration
For prompts with signed content:
```python theme={null}
# Already handled by get_prompt(), but if you need manual control:
for message in prompt.messages:
for block_group in message.blocks:
for block in block_group:
if isinstance(block, SignedURLContent):
await client.content_client.register_content(block)
```
## Image Variables
Use variables for dynamic images:
### In the Web App
1. Insert a variable with type "image"
2. At runtime, provide the image object
### In Code
Image variables use `ImageContentFromSource`, a **discriminated union** type. The SDK automatically dispatches to the correct image type based on the `type` field.
```python theme={null}
from moxn.base_models.blocks.image import ImageContentFromSource
from moxn.types.base import RenderableModel
class ImageAnalysisInput(RenderableModel):
query: str
screenshot: ImageContentFromSource
def render(self, **kwargs) -> dict:
return {
"query": self.query,
"screenshot": self.screenshot, # Returns the image object directly
}
```
### Constructing Images
You can provide images in two ways:
Pass a dict with the right structure—no extra imports needed:
```python theme={null}
# From URL
image = {
"type": "url",
"url": "https://example.com/screenshot.png",
"media_type": "image/png"
}
# From base64
image = {
"type": "base64",
"base64": "iVBORw0KGgo...",
"media_type": "image/png"
}
session_data = ImageAnalysisInput(
query="What's in this image?",
screenshot=image
)
```
Import and use the specific type for better IDE support:
```python theme={null}
from moxn.base_models.blocks.image import MediaImageFromURL
image = MediaImageFromURL(
url="https://example.com/screenshot.png",
media_type="image/png"
)
session_data = ImageAnalysisInput(
query="What's in this image?",
screenshot=image
)
```
Both approaches are equivalent—the dict is validated and converted using Pydantic's discriminated union dispatch.
### Image Types
| `type` value | Class | Required Fields |
| -------------- | ------------------------- | ------------------------ |
| `"url"` | `MediaImageFromURL` | `url`, `media_type` |
| `"base64"` | `MediaImageFromBase64` | `base64`, `media_type` |
| `"bytes"` | `MediaImageFromBytes` | `bytes`, `media_type` |
| `"local_file"` | `MediaImageFromLocalFile` | `filepath`, `media_type` |
Supported `media_type` values: `"image/jpeg"`, `"image/png"`, `"image/gif"`, `"image/webp"`
## Provider-Specific Handling
Moxn supports multiple provider APIs, each with different multimodal capabilities:
### Anthropic
```python theme={null}
session.to_anthropic_invocation()
```
* Images: PNG, JPEG, GIF, WebP
* PDFs: Native support with citations
### OpenAI
OpenAI has two distinct APIs with different invocation methods:
```python theme={null}
# Chat Completions API
session.to_openai_chat_invocation()
# Responses API (different format)
session.to_openai_responses_invocation()
```
* Images: Via data URIs or URLs
* PDFs: Limited support
### Google
Google has two distinct APIs:
```python theme={null}
# Gemini Developer API
session.to_google_gemini_invocation()
# Vertex AI (different authentication and endpoints)
session.to_google_vertex_invocation()
```
* Images: Various formats
* PDFs: Native support
* Note: Vertex AI requires GCS URIs (`gs://`) for remote files—public HTTP URLs are not supported
## Error Handling
Handle multimodal-specific errors:
```python theme={null}
try:
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
except anthropic.BadRequestError as e:
if "image" in str(e).lower():
# Image format or size issue
print(f"Image error: {e}")
raise
```
## Complete Example
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from moxn.base_models.blocks.image import ImageContentFromSource
from moxn.types.base import RenderableModel
from anthropic import Anthropic
class ImageAnalysisInput(RenderableModel):
"""Input with an image for analysis."""
screenshot: ImageContentFromSource
question: str
def render(self, **kwargs) -> dict:
return {
"screenshot": self.screenshot,
"question": self.question,
}
async def analyze_screenshot(image_url: str, question: str):
async with MoxnClient() as client:
# Construct image using dict literal (or use explicit MediaImageFromURL)
image = {"type": "url", "url": image_url, "media_type": "image/png"}
session = await client.create_prompt_session(
prompt_id="image-analysis-prompt",
session_data=ImageAnalysisInput(
screenshot=image,
question=question
)
)
async with client.span(
session,
name="analyze_image",
metadata={"has_image": True}
) as span:
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
return response.content[0].text
```
## Next Steps
Parse structured responses
Provider-specific handling
Build a document analysis pipeline
Log multimodal interactions
# Structured Outputs
Source: https://moxn.mintlify.app/advanced/structured-outputs
Get JSON responses that conform to a schema
Structured outputs let you get JSON responses that conform to a predefined schema. This guide covers how to configure and use structured outputs.
## What Are Structured Outputs?
Instead of free-form text:
```
The sentiment is positive with high confidence.
```
Get structured JSON:
```json theme={null}
{
"sentiment": "positive",
"confidence": 0.95,
"keywords": ["happy", "satisfied", "great"]
}
```
## Setting Up Structured Outputs
### In the Web App
Define the output structure in the Schemas section.
Mark the schema as "Structured Output" (not "Tool").
Attach the schema to your prompt's structured output setting.
### Schema Example
```json theme={null}
{
"type": "object",
"properties": {
"sentiment": {
"type": "string",
"enum": ["positive", "negative", "neutral"]
},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"keywords": {
"type": "array",
"items": {"type": "string"}
},
"reasoning": {
"type": "string"
}
},
"required": ["sentiment", "confidence"]
}
```
## Provider Support
| Provider | Structured Output Support |
| --------- | ------------------------- |
| Anthropic | Yes (beta) |
| OpenAI | Yes |
| Google | Yes |
## Using in Code
### Basic Usage
```python theme={null}
session = await client.create_prompt_session(
prompt_id="sentiment-analysis-prompt",
session_data=AnalysisInput(text="Great product!")
)
# Structured output schema is included automatically
response = anthropic.messages.create(
**session.to_anthropic_invocation(),
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
)
```
### Parsing the Response
```python theme={null}
import json
from pydantic import BaseModel
class SentimentResult(BaseModel):
sentiment: str
confidence: float
keywords: list[str] = []
reasoning: str | None = None
# Parse the response (uses stored provider from completion_config)
parsed = session.parse_response(response)
text_content = parsed.candidates[0].content[0].text
# Validate with Pydantic
result = SentimentResult.model_validate_json(text_content)
print(f"Sentiment: {result.sentiment} ({result.confidence:.0%})")
```
## Provider-Specific Handling
### Anthropic
Requires beta header:
```python theme={null}
response = anthropic.messages.create(
**session.to_anthropic_invocation(),
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
)
```
Generated payload includes:
```python theme={null}
{
"output_format": {
"type": "json_schema",
"json_schema": {
"name": "SentimentResult",
"schema": {...}
}
}
}
```
### OpenAI
Works out of the box:
```python theme={null}
response = openai.chat.completions.create(
**session.to_openai_chat_invocation()
)
```
Generated payload includes:
```python theme={null}
{
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "SentimentResult",
"schema": {...},
"strict": True
}
}
}
```
### Google
```python theme={null}
response = client.models.generate_content(
**session.to_google_gemini_invocation()
)
```
Generated config includes:
```python theme={null}
{
"config": {
"response_schema": {...},
"response_mime_type": "application/json"
}
}
```
## Validation Patterns
### Basic Validation
```python theme={null}
from pydantic import ValidationError
try:
result = SentimentResult.model_validate_json(response_text)
except ValidationError as e:
print(f"Invalid response: {e}")
```
### With Telemetry
```python theme={null}
validation_errors = []
try:
result = SentimentResult.model_validate_json(response_text)
except ValidationError as e:
validation_errors = [str(err) for err in e.errors()]
result = None
# Log with validation status
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
validation_errors=validation_errors or None
)
await client.log_telemetry_event(event)
```
### Retry on Failure
```python theme={null}
async def analyze_with_retry(text: str, max_retries: int = 2):
for attempt in range(max_retries + 1):
response = await call_llm()
parsed = session.parse_response(response)
try:
result = SentimentResult.model_validate_json(
parsed.candidates[0].content[0].text
)
return result
except ValidationError as e:
if attempt == max_retries:
raise
# Optionally: add error to context for next attempt
```
## Schema Design Tips
### Use Enums for Categories
```json theme={null}
{
"category": {
"type": "string",
"enum": ["bug", "feature", "question", "other"]
}
}
```
### Set Bounds for Numbers
```json theme={null}
{
"score": {
"type": "number",
"minimum": 0,
"maximum": 10
}
}
```
### Use Required Fields
```json theme={null}
{
"required": ["category", "confidence"]
}
```
### Allow Additional Context
Include a reasoning field for explainability:
```json theme={null}
{
"reasoning": {
"type": "string",
"description": "Explanation for the classification"
}
}
```
## Tools vs Structured Outputs
| Aspect | Tools (Function Calling) | Structured Outputs |
| ---------------- | ------------------------- | ------------------ |
| Use when | LLM decides to use a tool | Always want JSON |
| Control | LLM chooses when | Guaranteed |
| Response | May include tool calls | Pure JSON |
| Multiple schemas | Yes (multiple tools) | One schema |
**Use tools when:** The LLM should decide whether to call a function
**Use structured output when:** You always want a specific JSON format
## Combining with Tools
You can have both tools and structured output:
* Tools for actions (search, calculate, etc.)
* Structured output for the final response format
```python theme={null}
# Prompt has both tools and structured output configured
response = anthropic.messages.create(
**session.to_anthropic_invocation(),
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
)
# Response may include tool calls followed by structured output
for block in response.content:
if block.type == "tool_use":
# Handle tool call
result = execute_tool(block.name, block.input)
elif block.type == "text":
# Parse structured output
output = SentimentResult.model_validate_json(block.text)
```
## Complete Example
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
from pydantic import BaseModel, ValidationError
class ExtractionResult(BaseModel):
entities: list[str]
summary: str
category: str
confidence: float
async def extract_from_document(text: str):
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="extraction-prompt",
session_data=ExtractionInput(document=text)
)
async with client.span(
session,
name="extract_entities",
metadata={"doc_length": len(text)}
) as span:
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation(),
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
)
parsed = session.parse_response(response)
# Validate structured output
validation_errors = []
try:
result = ExtractionResult.model_validate_json(
parsed.candidates[0].content[0].text
)
except ValidationError as e:
validation_errors = [str(err) for err in e.errors()]
result = None
# Log with validation status in event attributes
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
validation_errors=validation_errors or None,
attributes={
"extraction_success": result is not None,
"entity_count": len(result.entities) if result else 0
}
)
await client.log_telemetry_event(event)
return result
```
## Next Steps
Complete extraction example
Provider-specific details
Generate models for outputs
Log validation results
# Entity Model
Source: https://moxn.mintlify.app/concepts/entities
Understanding Tasks, Prompts, Messages, and Schemas
Moxn organizes content in a hierarchical structure designed for collaboration and reuse. This guide explains each entity type and how they relate.
## Entity Hierarchy
```mermaid theme={null}
graph TD
Task[Task - Repository]
Task --> Prompts[Prompts - LLM Templates]
Task --> Schemas[Schemas - Tool/Output Definitions]
Task --> Version[Branches & Commits - Version Control]
Prompts --> Messages[Messages - Content Blocks]
Prompts --> InputSchema[Input Schema - Auto-generated]
Messages --> Variables[Variables → Properties]
Schemas --> Properties[Properties - Type Definitions]
```
## Task
A **Task** is the top-level container, analogous to a Git repository. It groups related prompts and provides version control boundaries.
```python theme={null}
task = await client.get_task("task-id", branch_name="main")
task.id # UUID - stable identifier
task.name # str - e.g., "Customer Support Bot"
task.description # str | None
task.prompts # list[PromptTemplate]
task.definitions # dict[str, SchemaDefinition] - all schemas
task.branches # list[Branch]
task.last_commit # Commit | None
```
**Use cases for Tasks:**
* One task per AI feature (customer support, search, etc.)
* One task per team or domain
* One task per environment (if you prefer separate staging/production)
## Prompt
A **Prompt** (or PromptTemplate) is a template for a single LLM invocation. It combines messages, an input schema, and optional tool definitions.
```python theme={null}
prompt = await client.get_prompt("prompt-id", branch_name="main")
prompt.id # UUID
prompt.name # str - e.g., "Product Help"
prompt.description # str | None
prompt.task_id # UUID - parent task
prompt.messages # list[Message]
prompt.input_schema # Schema | None - auto-generated from variables
prompt.tools # list[SdkTool] - function calling tools
prompt.completion_config # CompletionConfig - model settings
```
### Completion Config
Each prompt can specify default model settings:
```python theme={null}
prompt.completion_config.provider # Provider enum
prompt.completion_config.model # str - e.g., "claude-sonnet-4-20250514"
prompt.completion_config.max_tokens # int
prompt.completion_config.temperature # float
prompt.completion_config.tool_choice # str - "auto", "required", "none", or tool name
```
### Tools
Prompts can have tools for function calling or structured output:
```python theme={null}
# Function calling tools
prompt.function_tools # list[SdkTool] where tool_type == "tool"
# Structured output schema
prompt.structured_output_schema # SdkTool | None where tool_type == "structured_output"
```
## Message
A **Message** is a reusable content block with a specific role. Messages can be shared across multiple prompts.
```python theme={null}
for message in prompt.messages:
message.id # UUID
message.name # str - e.g., "System Prompt"
message.role # "system" | "user" | "assistant"
message.author # "HUMAN" | "MACHINE"
message.blocks # list[list[ContentBlock]] - 2D content
message.task_id # UUID - parent task
```
### Message Roles
| Role | Description |
| ----------- | ------------------------------------------------------------------------- |
| `system` | Instructions for the LLM (converted to system param for Anthropic/Google) |
| `user` | User input or context |
| `assistant` | Assistant responses (for few-shot examples or prefilling) |
### Content Blocks
Messages contain **blocks** in a 2D array structure:
```python theme={null}
message.blocks # list[list[ContentBlock]]
# Outer array: paragraphs or logical sections
# Inner array: blocks within a paragraph
```
**Block types:**
```python theme={null}
# Text
TextContent(text="You are a helpful assistant.")
# Variables (substituted at runtime)
Variable(
name="query",
variable_type="primitive", # or "image", "file"
format="inline", # or "block"
description="The user's question",
required=True
)
# Images
ImageContentFromSource(url="https://...", alt="description")
SignedURLImageContent(signed_url="...", file_path="...", expiration=datetime)
# PDFs
PDFContentFromSource(url="https://...", name="document.pdf")
SignedURLPDFContent(signed_url="...", file_path="...", expiration=datetime)
# Tool interactions
ToolCall(tool_name="search", tool_call_id="...", input={...})
ToolResult(tool_call_id="...", content="...")
# Extended thinking
ThinkingContent(text="Let me reason about this...")
ReasoningContent(text="Step 1: ...")
```
## Property
A **Property** is a type definition that describes a variable. Properties define the shape of data that flows into prompts.
Properties are composed into schemas and linked to variables in messages by name.
**Property definition (JSON Schema):**
```json theme={null}
{
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The user's question"
},
"context": {
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string"}
}
}
}
},
"required": ["query"]
}
```
**Supported types:**
* Primitives: `string`, `number`, `integer`, `boolean`
* Complex: `object`, `array`
* Formats: `date`, `date-time`, `email`, `uri`, `uuid`
## Schema
A **Schema** is a compiled JSON Schema definition, either:
* **Input schema**: Auto-generated from variables in a prompt's messages
* **Tool schema**: Manually created for function calling or structured output
```python theme={null}
schema = prompt.input_schema
schema.id # UUID
schema.name # str - e.g., "ProductHelpInput"
schema.description # str | None
schema.exportedJSON # str - JSON Schema as string
```
### Input Schema Auto-Sync
When you add variables to messages, the input schema updates automatically:
```mermaid theme={null}
flowchart TB
A["Message: Answer the question: {query}"] --> B["Variable linked to Property 'query' (type: string)"]
B --> C["Input Schema gains 'query' property automatically"]
```
This keeps your schema in sync with your actual prompt content.
### Schema Definitions
Access full schema definitions via the task:
```python theme={null}
task = await client.get_task("task-id")
for name, definition in task.definitions.items():
print(f"Schema: {name}")
print(f" Type: {definition['type']}")
print(f" Properties: {list(definition.get('properties', {}).keys())}")
# Moxn metadata embedded in schema
metadata = definition.get("x-moxn-metadata", {})
print(f" Schema ID: {metadata.get('schema_id')}")
print(f" Prompt ID: {metadata.get('prompt_id')}")
```
## Relationships
### Pass-by-Reference
Messages are **referenced**, not copied, across prompts:
```mermaid theme={null}
graph TD
Task --> PromptA[Prompt A]
Task --> PromptB[Prompt B]
Task --> MsgPool[Messages]
PromptA --> |uses| msg1[msg_1 - System shared]
PromptA --> |uses| msg2[msg_2]
PromptA --> |uses| msg3[msg_3]
PromptB --> |uses| msg1
PromptB --> |uses| msg4[msg_4]
```
This means:
* Updating a shared message updates all prompts using it
* Each prompt still gets its own copy at fetch time (SDK doesn't maintain live references)
* Reuse is tracked in the web app for easier management
### Anchor IDs
Every entity has a stable **anchor ID** that never changes:
```python theme={null}
prompt.id # UUID - stable across all branches and commits
```
When you fetch a prompt from different branches, the ID stays the same. Only the content changes.
This enables:
* Referencing prompts in code without worrying about versions
* Tracking the same entity across branches
* Stable foreign key relationships
## Entity Lifecycle
```mermaid theme={null}
flowchart TB
Create[Create - Web App] --> Edit[Edit - Working State]
Edit --> Commit[Commit - Version Record]
Commit --> Access[Access - SDK]
Access --> Branch[By branch → Gets working state]
Access --> CommitAccess[By commit → Gets version record - immutable]
```
See [Versioning](/concepts/versioning) for details on branches and commits.
## Next Steps
Understand branches and commits
Deep dive into the variable system
Fetch and use prompts in code
Create entities in the web app
# Variables & Schemas
Source: https://moxn.mintlify.app/concepts/variables
How variables, properties, and schemas work together
Variables are the bridge between your application data and your prompts. This guide explains how the variable system works.
## The Variable System
```mermaid theme={null}
flowchart LR
A["Your Code
(Pydantic Model)"] -->|"render()"| B["Moxn
(dict[str, str])"]
B -->|"substitute"| C["LLM
(Final Prompt)"]
```
## How Variables Work
### 1. Define Variables in Messages
In the web app, insert variables using the `/variable` command:
```
System: You are a customer support agent for {{company_name}}.
User: Customer {{customer_name}} asks:
{{query}}
Relevant context:
{{search_results}}
```
Each `{{variable_name}}` links to a **Property** that defines its type.
### 2. Variables Sync to Schema
When you add variables, the prompt's **input schema** updates automatically:
```json theme={null}
{
"type": "object",
"properties": {
"company_name": {"type": "string"},
"customer_name": {"type": "string"},
"query": {"type": "string"},
"search_results": {"type": "string"}
},
"required": ["query"]
}
```
This happens in real-time as you edit—no manual schema management.
### 3. Provide Data at Runtime
At runtime, provide values that match the schema:
```python theme={null}
session = await client.create_prompt_session(
prompt_id="...",
session_data=ProductHelpInput(
company_name="Acme Corp",
customer_name="Alice",
query="How do I reset my password?",
search_results="[relevant docs as JSON]"
)
)
```
### 4. Variables Get Substituted
When converting to provider format, variables are replaced:
```python theme={null}
# Original message block
"Customer {{customer_name}} asks: {{query}}"
# After substitution
"Customer Alice asks: How do I reset my password?"
```
## Properties
A **Property** defines the type and metadata for a variable.
### Property Types
| Type | JSON Schema | Python | Use Case |
| ------- | ------------------------- | ------------ | --------------- |
| String | `{"type": "string"}` | `str` | Text, IDs |
| Number | `{"type": "number"}` | `float` | Scores, prices |
| Integer | `{"type": "integer"}` | `int` | Counts |
| Boolean | `{"type": "boolean"}` | `bool` | Flags |
| Object | `{"type": "object", ...}` | Nested model | Structured data |
| Array | `{"type": "array", ...}` | `list[T]` | Collections |
### Special Formats
String properties can have formats:
```json theme={null}
{
"type": "string",
"format": "date" // "2024-01-15"
}
{
"type": "string",
"format": "date-time" // "2024-01-15T10:30:00Z"
}
{
"type": "string",
"format": "email" // "user@example.com"
}
{
"type": "string",
"format": "uri" // "https://example.com"
}
```
### Complex Properties
Properties can define nested structures:
```json theme={null}
{
"type": "object",
"properties": {
"id": {"type": "string"},
"title": {"type": "string"},
"content": {"type": "string"},
"score": {"type": "number"}
},
"required": ["id", "title", "content"]
}
```
Or arrays of complex objects:
```json theme={null}
{
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string"}
}
}
}
```
## The render() Transformation
### Why render()?
Variables must be **strings** for prompt injection, but your code uses **typed data**.
The `render()` method bridges this gap:
```python theme={null}
class ProductHelpInput(RenderableModel):
customer_name: str # Simple: just use as-is
query: str
documents: list[Document] # Complex: needs serialization
def render(self, **kwargs) -> dict[str, str]:
return {
"customer_name": self.customer_name,
"query": self.query,
"documents": json.dumps([d.model_dump() for d in self.documents])
}
```
### render() Input vs Output
```python theme={null}
# Input (typed)
input = ProductHelpInput(
customer_name="Alice",
query="How do I...",
documents=[
Document(title="FAQ", content="..."),
Document(title="Guide", content="...")
]
)
# Output (all strings)
output = input.render()
# {
# "customer_name": "Alice",
# "query": "How do I...",
# "documents": '[{"title": "FAQ", ...}, {"title": "Guide", ...}]'
# }
```
### Customizing Format
Override `render()` to control formatting:
```python theme={null}
def render(self, **kwargs) -> dict[str, str]:
# Markdown format
docs_md = "\n".join([
f"## {d.title}\n{d.content}"
for d in self.documents
])
return {
"customer_name": self.customer_name,
"query": self.query,
"documents": docs_md
}
```
Or XML:
```python theme={null}
def render(self, **kwargs) -> dict[str, str]:
docs_xml = "\n" + "\n".join([
f' {d.content}'
for d in self.documents
]) + "\n"
return {
"customer_name": self.customer_name,
"query": self.query,
"documents": docs_xml
}
```
## Schemas
### Input Schemas
Every prompt has an **input schema** that defines required variables:
```python theme={null}
prompt = await client.get_prompt("...", branch_name="main")
# Access the input schema
schema = prompt.input_schema
print(schema.name) # "ProductHelpInput"
print(schema.exportedJSON) # Full JSON Schema
```
Input schemas are **auto-generated** from message variables. You don't create them manually.
### Tool Schemas
Tool schemas define function calling or structured output:
```python theme={null}
# Access tool schemas
for tool in prompt.tools:
if tool.tool_type == "tool":
print(f"Function: {tool.name}")
print(f"Schema: {tool.schema}")
elif tool.tool_type == "structured_output":
print(f"Output schema: {tool.name}")
```
Tool schemas are created manually in the web app.
## Variable Substitution
### How It Works
```mermaid theme={null}
flowchart TB
A["1. Session created with session_data"] --> A1["ProductHelpInput(customer_name='Alice', ...)"]
A1 --> B["2. render() called on session_data"]
B --> B1["{'customer_name': 'Alice', ...}"]
B1 --> C["3. Each message block processed"]
C --> C1["'Hello {{customer_name}}' → 'Hello Alice'"]
C1 --> D["4. Result sent to LLM"]
```
### Variable Formats
Variables can be **inline** or **block**:
| Format | Example | Description |
| ---------- | ---------------------------------------- | ------------------------- |
| **Inline** | `The customer {{customer_name}} said...` | Variable embedded in text |
| **Block** | `{{search_results}}` | Variable is entire block |
This affects rendering—block variables typically appear on their own line.
## Schema Metadata
Schemas include Moxn metadata:
```json theme={null}
{
"type": "object",
"properties": {...},
"x-moxn-metadata": {
"schema_id": "uuid",
"prompt_id": "uuid",
"task_id": "uuid",
"commit_id": "abc123" // If from a commit
}
}
```
Codegen'd models expose this:
```python theme={null}
class ProductHelpInput(RenderableModel):
@classmethod
@property
def moxn_schema_metadata(cls) -> MoxnSchemaMetadata:
return MoxnSchemaMetadata(
schema_id="...",
prompt_id="...",
task_id="..."
)
```
## Best Practices
`customer_query` is better than `q`. Variable names appear in code and logs.
Use specific types (date, integer) rather than just string when possible.
Format complex data (markdown, XML) for better LLM comprehension.
Each prompt should have just the variables it needs.
When the number of items varies, use array types.
## Next Steps
Generate models from schemas
Define variables in the editor
Use variables at runtime
Understand the full data model
# Versioning
Source: https://moxn.mintlify.app/concepts/versioning
Git-like version control for prompts
Moxn implements a Git-inspired versioning system. This guide explains how branches, commits, and working state work together.
## Core Concepts
### Branches
A **branch** is an isolated workspace for changes:
```mermaid theme={null}
graph LR
subgraph main [main - default branch]
m1[System prompt v1]
m2[User template v1]
m3[Working changes...]
end
subgraph feature [feature-improvements]
f1[System prompt v2 - modified]
f2[User template v1 - same]
f3[More changes...]
end
```
**Key properties:**
* Each task has a default branch (usually "main")
* Branches are isolated—changes don't affect other branches
* Create branches for experimentation without risk
### Commits
A **commit** is an immutable snapshot of the entire task state:
```mermaid theme={null}
graph TD
C[Commit: abc123]
C --> T["Task: name='Support Bot', description='...'"]
C --> P["Prompts: [prompt_1_v5, prompt_2_v3]"]
C --> M["Messages: [msg_1_v2, msg_2_v1, msg_3_v1]"]
C --> S["Schemas: [schema_1_v2]"]
```
**Key properties:**
* Commits are immutable—once created, they never change
* A commit captures ALL entities in the task at that moment
* Use commit IDs for reproducibility
### Working State
**Working state** represents uncommitted changes on a branch:
```mermaid theme={null}
graph TD
B[Branch: feature-xyz]
B --> H[Head Commit: abc123]
B --> W[Working State]
W --> WM["Modified: message_2 (new content)"]
W --> WA[Added: message_4]
W --> WD[Deleted: message_3]
```
**Key properties:**
* Working state is mutable
* Only exists per branch
* Gets cleared when you commit
## Access Patterns
### Branch Access
Fetching by branch returns the **latest state** (including uncommitted changes):
```python theme={null}
# Gets working state + base commit
prompt = await client.get_prompt(
prompt_id="...",
branch_name="main"
)
```
**What you get:**
1. Start with the branch's head commit
2. Overlay any working state modifications
3. Apply any deletions
4. Return the merged "current state"
**Use for:** Development, testing, rapid iteration
### Commit Access
Fetching by commit returns an **immutable snapshot**:
```python theme={null}
# Gets exact state at commit time
prompt = await client.get_prompt(
prompt_id="...",
commit_id="abc123def456"
)
```
**What you get:**
* The exact prompt state when that commit was created
* Guaranteed to never change
* Cached indefinitely by the SDK
**Use for:** Production, reproducibility, auditing
### Branch Head Access
To get the latest **committed** state (without working changes):
```python theme={null}
# Step 1: Get the branch head
head = await client.get_branch_head(
task_id="task-uuid",
branch_name="main"
)
# Step 2: Fetch by that commit
prompt = await client.get_prompt(
prompt_id="...",
commit_id=head.effectiveCommitId
)
```
**What you get:**
* The most recent committed state
* No work-in-progress changes
* Still immutable
**Use for:** Deploying "stable" versions, CI/CD
## Resolution Logic
When you fetch by branch, here's what happens:
```mermaid theme={null}
flowchart TB
A["1. Find branch head commit"] --> A1["commit 'abc123'"]
A1 --> B["2. Load version records from commit"]
B --> B1["prompt_v5, message_v2, message_v1"]
B1 --> C["3. Check for working state modifications"]
C --> C1["message_v2 has working state changes"]
C1 --> D["4. Check for branch deletions"]
D --> D1["message_3 was deleted"]
D1 --> E["5. Merge to current state"]
E --> E1["prompt with modified message_2, without message_3"]
```
This resolution happens server-side. The SDK receives the resolved state.
## Production Deployment Patterns
### Pattern 1: Pin to Commit
The safest approach—pin production to a specific commit:
```python theme={null}
# In your config or environment
PROMPT_COMMIT_ID = "abc123def456"
# In your code
session = await client.create_prompt_session(
prompt_id="...",
commit_id=PROMPT_COMMIT_ID
)
```
**Pros:**
* Completely immutable
* Perfect reproducibility
* Safe from accidental changes
**Cons:**
* Manual updates required
* Need to track commit IDs
### Pattern 2: Latest Commit on Branch
Use the branch head, but only committed state:
```python theme={null}
# Get latest committed state
head = await client.get_branch_head("task-id", "production")
session = await client.create_prompt_session(
prompt_id="...",
commit_id=head.effectiveCommitId
)
```
**Pros:**
* Automatically gets new commits
* No work-in-progress changes
* Can use a "production" branch for controlled releases
**Cons:**
* Any commit to the branch goes live
* Need branch discipline
### Pattern 3: Branch Access (Development Only)
For development and testing:
```python theme={null}
session = await client.create_prompt_session(
prompt_id="...",
branch_name="main"
)
```
**Pros:**
* Always latest, including uncommitted
* Great for rapid iteration
**Cons:**
* Can change unexpectedly
* Not suitable for production
## Caching Behavior
The SDK caches based on access pattern:
| Access | Cached | Reason |
| ------------------- | ------ | --------------------- |
| `commit_id="..."` | Yes | Commits are immutable |
| `branch_name="..."` | No | Branches change |
```python theme={null}
# First call: fetches from API
prompt1 = await client.get_prompt("...", commit_id="abc123")
# Second call: returns from cache
prompt2 = await client.get_prompt("...", commit_id="abc123")
# Always fetches: branch could have changed
prompt3 = await client.get_prompt("...", branch_name="main")
```
## Branch Response Fields
When fetching a branch head:
```python theme={null}
head = await client.get_branch_head("task-id", "main")
head.branchId # UUID
head.branchName # str
head.taskId # UUID
head.headCommitId # str | None - latest commit on this branch
head.parentCommitId # str | None - where branch forked from
head.effectiveCommitId # str - what to use (head or parent if no commits)
head.hasUncommittedChanges # bool - is there working state?
head.lastCommittedAt # datetime | None
head.isDefault # bool - is this the default branch?
```
## Entity Version Context
Fetched entities include version context:
```python theme={null}
prompt = await client.get_prompt("...", branch_name="main")
prompt.branch_id # UUID | None - if fetched by branch
prompt.commit_id # UUID | None - if fetched by commit
```
This helps with:
* Knowing which version you're working with
* Including version info in telemetry
* Debugging version-related issues
## Immutability Guarantees
### What's Immutable
* **Commits**: Once created, never change
* **Version records**: The state captured in a commit
* **Content records**: Message and property content is content-addressed
### What's Mutable
* **Working state**: Changes until committed
* **Branch pointers**: Head moves with new commits
* **Deletions**: Tracked per-branch
## Best Practices
Always pin production deployments to specific commit IDs for reproducibility.
Don't experiment on main. Create branches to test changes safely.
Commits are cheap and create restore points.
During development, use branch access for rapid iteration.
Use a dedicated "production" branch that only gets stable commits.
## Next Steps
Understand the data model
Fetch prompts by branch or commit
Manage branches in the web app
Track which versions are used
# Why Moxn?
Source: https://moxn.mintlify.app/context
The problems Moxn solves for AI engineering teams
# Why Moxn?
Prompts are code, but they don't have the tooling.
When building AI applications, you face a familiar set of challenges:
* **No dedicated tooling**: Prompts are scattered across strings, YAML files, and spreadsheets—with no proper editor for structured content
* **Version control friction**: Prompts either live in git (coupled to deploys, painful for domain experts) or in config systems (schemaless, painful for engineers)
* **No type safety**: Input variables are *stringly*-typed, leading to runtime errors
* **No observability**: You can't see what prompts actually ran in production
* **No reuse**: Want a shared system message across multiple agent surfaces? Good luck
* **No collaboration**: No shared workflows for reviewing prompts or debugging production traces
Moxn solves these problems by treating prompts as first-class versioned entities with type-safe interfaces and full observability.
## Core Architecture
Moxn separates **content management** from **runtime execution**:
Messages, variables, and model config—stored in Moxn, versioned like code.
Template + your runtime data, created in your application.
A plain Python dict you pass directly to the provider SDK.
```mermaid theme={null}
sequenceDiagram
participant App as Your App
participant Moxn as Moxn SDK
participant LLM as Anthropic/OpenAI
App->>Moxn: get_prompt(prompt_id)
Moxn-->>App: Prompt Template
Note right of App: Template + session_data
App->>Moxn: create_prompt_session(template, data)
Moxn-->>App: Prompt Session
App->>App: session.to_anthropic_invocation()
Note right of App: Returns a plain dict
App->>LLM: messages.create(**invocation)
LLM-->>App: response
App->>Moxn: log_telemetry_event(response)
```
## Design Philosophy
### Moxn Builds Payloads, You Own the Integration
The SDK produces standard Python dictionaries. You unpack them directly into provider SDKs:
```python theme={null}
# Moxn builds the payload from your template + data
invocation = session.to_anthropic_invocation()
# You call the provider directly—no wrapper, no magic
response = anthropic.messages.create(**invocation)
```
This means you can always:
* **Modify the payload** before sending (add headers, override settings)
* **Use new provider features** without waiting for SDK updates
* **Compose with other tools** in your stack
Moxn simplifies the common path without making edge cases harder.
## Key Features
A block-based editor with mermaid diagrams, code blocks, XML documents, and
multimodal content. Author prompts and review traces in the same interface.
Branch, commit, and rollback prompts. Pin production to specific commits.
Review diffs before deploying.
Auto-generated Pydantic models ensure type safety from editor to runtime.
Get autocomplete and validation.
View traces and spans in the same rich editor. W3C Trace Context compatible
with complete LLM event logging.
![Message editor with reminders and code blocks]()
## Next Steps
Install and run your first prompt
See the complete development cycle
# Document Analysis Pipeline
Source: https://moxn.mintlify.app/examples/building
Step-by-step guide to creating prompts, variables, and schemas
This example walks through building a complete document analysis pipeline in the Moxn web app, demonstrating variables, typed properties, enums, and code generation.
## What We'll Build
A three-prompt pipeline that:
1. **Classifies** documents into categories (contract, invoice, report, etc.)
2. **Extracts** structured entities from the document
3. **Generates** a summary report
## Step 1: Create a Task
Navigate to the dashboard and click **Create Task**. Name it `document-analysis-pipeline` with a description.
A Task is a container for related prompts—like a Git repository for your AI features. All prompts within a task share schemas and are versioned together.
## Step 2: Create the Classifier Prompt
### Add a System Message
Create a new prompt called `document-classifier`. Add a system message with instructions:
```
You are a document classification expert. Analyze the provided document and classify it into one of the following categories: contract, invoice, report, memo, or email.
Be precise and consider the document's structure, language, and purpose.
```
### Add a User Message with Variables
Create a user message. Instead of hardcoding content, we'll use a **variable** to inject the document at runtime.
Type `/` in the editor to open the slash command menu:
![Slash command menu]()
Select **Variable Block** to open the property editor:
![Property editor]()
Configure the variable:
* **Property Name**: `document`
* **Description**: "The document content to classify"
* **Type**: String
The type dropdown shows all available types:
![Type dropdown]()
After clicking **Create**, the variable appears as a styled block in your message:
![Variable block in editor]()
## Step 3: Define the Output Schema
Navigate to the **Schemas** tab. You'll see:
* **Input Schemas**: Auto-generated from prompt variables
* **User-Defined Schemas**: Custom schemas for structured outputs
![Schemas overview]()
### Create an Enum Schema
Click **Create Schema** and name it `ClassificationResult`. Add a property `document_type` with **allowed values** to create an enum:
![Enum configuration]()
Enter comma-separated values: `contract, invoice, report, memo, email`
The validation message confirms: "Must be one of: contract, invoice, report, memo, email"
## Step 4: Generate Pydantic Models
Run code generation to create typed Python models:
```python theme={null}
import asyncio
from moxn import MoxnClient
async def generate():
async with MoxnClient() as client:
await client.generate_task_models(
task_id="your-task-id",
branch_name="main",
output_dir="./models"
)
asyncio.run(generate())
```
This generates:
```python theme={null}
# models/document_analysis_pipeline_models.py
from enum import Enum
from pydantic import BaseModel
from moxn.types.base import RenderableModel
class DocumentType(str, Enum):
CONTRACT = "contract"
INVOICE = "invoice"
REPORT = "report"
MEMO = "memo"
EMAIL = "email"
class ClassificationResult(BaseModel):
"""Structured output schema for document classification."""
document_type: DocumentType
confidence: float | None = None
class DocumentClassifierInput(RenderableModel):
"""Input schema for document-classifier prompt."""
document: str
def render(self, **kwargs) -> dict[str, str]:
return {"document": self.document}
```
## Step 5: Use in Your Application
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
from models.document_analysis_pipeline_models import (
DocumentClassifierInput,
ClassificationResult
)
async def classify_document(document_text: str) -> ClassificationResult:
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="document-classifier",
branch_name="main",
session_data=DocumentClassifierInput(document=document_text)
)
async with client.span(
session,
name="classify_document",
metadata={"doc_length": len(document_text)}
) as span:
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation(),
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
)
parsed = session.parse_response(response)
result = ClassificationResult.model_validate_json(
parsed.candidates[0].content[0].text
)
# Log with classification result in event attributes
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
attributes={"document_type": result.document_type.value}
)
await client.log_telemetry_event(event)
return result
```
## Extending the Pipeline
### Entity Extractor with Object Schema
Create a second prompt `entity-extractor` with an object schema for extracted entities:
```python theme={null}
class ExtractedEntities(BaseModel):
"""Entities extracted from a document."""
people: list[str] = []
organizations: list[str] = []
dates: list[str] = []
amounts: list[float] = []
key_terms: list[str] = []
class EntityExtractorInput(RenderableModel):
document: str
document_type: DocumentType # Reference the enum from classifier
def render(self, **kwargs) -> dict[str, str]:
return {
"document": self.document,
"document_type": self.document_type.value
}
```
### Report Generator with Schema Reference
Create a third prompt `report-generator` that references outputs from earlier prompts:
```python theme={null}
class ReportGeneratorInput(RenderableModel):
document: str
classification: ClassificationResult # Reference classifier output
entities: ExtractedEntities # Reference extractor output
def render(self, **kwargs) -> dict[str, str]:
return {
"document": self.document,
"classification": self.classification.model_dump_json(),
"entities": self.entities.model_dump_json()
}
```
## Complete Pipeline
```python theme={null}
async def analyze_document(document_text: str) -> dict:
"""Run the complete document analysis pipeline."""
async with MoxnClient() as client:
# Step 1: Classify
classification = await classify_document(document_text)
# Step 2: Extract entities
entities = await extract_entities(
document_text,
classification.document_type
)
# Step 3: Generate report
report = await generate_report(
document_text,
classification,
entities
)
return {
"classification": classification,
"entities": entities,
"report": report
}
```
## Key Concepts Demonstrated
| Feature | Where Used |
| --------------------- | ------------------------------------------ |
| **Variables** | `document` variable in classifier input |
| **Enums** | `DocumentType` with allowed values |
| **Objects** | `ExtractedEntities` with nested fields |
| **Schema References** | `ReportGeneratorInput` using other schemas |
| **Code Generation** | Type-safe Pydantic models |
| **Telemetry** | Spans and logging for observability |
## Next Steps
Add retrieval to your prompts
Parallel execution patterns
Advanced variable configuration
Schema design patterns
# Code Generation
Source: https://moxn.mintlify.app/guides/codegen
Generate type-safe Pydantic models from your prompts
Moxn can generate Pydantic models from your prompt schemas, giving you type-safe interfaces for your LLM applications. This guide covers how code generation works and how to use the generated models.
## Why Code Generation?
Without codegen, you'd write:
```python theme={null}
# No type safety, easy to make mistakes
session = await client.create_prompt_session(
prompt_id="...",
session_data={"qurey": "typo here", "user_id": 123} # Oops!
)
```
With codegen:
```python theme={null}
# Type-safe, IDE autocomplete, validation
from generated_models import ProductHelpInput
session = await client.create_prompt_session(
prompt_id="...",
session_data=ProductHelpInput(
query="How do I reset my password?", # Autocomplete!
user_id="user_123" # Type checked!
)
)
```
## Generating Models
### Using MoxnClient
Generate models for all prompts in a task:
```python theme={null}
from moxn import MoxnClient
async with MoxnClient() as client:
result = await client.generate_task_models(
task_id="your-task-id",
branch_name="main", # or commit_id="..."
output_dir="./generated" # Where to save the file
)
print(f"Generated: {result.filename}")
print(f"Code:\n{result.generated_code}")
```
**Parameters:**
* `task_id`: The task containing your prompts
* `branch_name` or `commit_id`: Which version to generate from
* `output_dir`: Directory to write the generated file (optional)
**Returns:** `DatamodelCodegenResponse` with:
* `filename`: The generated file name
* `generated_code`: The Python code
### Output File
The generated file is named after your task:
* generated/
* customer\_support\_bot\_models.py
## What Gets Generated
For each prompt with an input schema, you get:
### 1. A Pydantic Model
```python theme={null}
from pydantic import Field
from moxn.types.base import RenderableModel, MoxnSchemaMetadata
class ProductHelpInput(RenderableModel):
"""Input schema for Product Help prompt."""
query: str = Field(..., description="The user's question")
user_id: str = Field(..., description="User identifier")
documents: list[Document] = Field(
default_factory=list,
description="Relevant documents from search"
)
@classmethod
@property
def moxn_schema_metadata(cls) -> MoxnSchemaMetadata:
return MoxnSchemaMetadata(
schema_id="...",
prompt_id="...",
task_id="..."
)
def render(self, **kwargs) -> dict[str, str]:
return {
"query": self.query,
"user_id": self.user_id,
"documents": json.dumps([d.model_dump() for d in self.documents]),
}
```
### 2. A TypedDict for Rendered Output
```python theme={null}
class ProductHelpInputRendered(TypedDict):
"""Rendered (string-valued) version of ProductHelpInput."""
query: str
user_id: str
documents: str # JSON string
```
### 3. Nested Types
Complex schemas generate nested models:
```python theme={null}
class Document(RenderableModel):
"""A search result document."""
id: str
title: str
content: str
score: float
class ProductHelpInput(RenderableModel):
documents: list[Document]
```
## The Two Representations
Code generation produces **two related types** for each schema:
| Type | Purpose | Values |
| -------------- | ------------------------------- | -------------------------- |
| Pydantic Model | Input validation, IDE support | Typed (str, int, list\[T]) |
| TypedDict | What gets injected into prompts | All strings |
This separation exists because:
1. **Your code** works with typed data (lists, numbers, nested objects)
2. **Prompts** receive string values (JSON, markdown, custom formats)
The `render()` method bridges these two:
```python theme={null}
# Input: typed data
input = ProductHelpInput(
query="How do I...",
documents=[Document(id="1", title="FAQ", content="...", score=0.95)]
)
# Output: flat string dict
rendered = input.render()
# {"query": "How do I...", "documents": "[{\"id\": \"1\", ...}]"}
```
## Customizing render()
The generated `render()` method provides a default implementation, but you can override it:
### Default behavior
```python theme={null}
def render(self, **kwargs) -> dict[str, str]:
return {
"query": self.query,
"documents": json.dumps([d.model_dump() for d in self.documents])
}
```
### Custom markdown formatting
```python theme={null}
class ProductHelpInput(RenderableModel):
documents: list[Document]
def render(self, **kwargs) -> dict[str, str]:
# Format documents as markdown
docs_md = "\n\n".join([
f"## {doc.title}\n\n{doc.content}\n\n*Relevance: {doc.score:.0%}*"
for doc in self.documents
])
return {
"query": self.query,
"documents": docs_md
}
```
### Custom XML formatting
```python theme={null}
def render(self, **kwargs) -> dict[str, str]:
docs_xml = "\n".join([
f'\n'
f' {doc.title}\n'
f' {doc.content}\n'
f''
for doc in self.documents
])
return {
"query": self.query,
"documents": f"\n{docs_xml}\n"
}
```
### Using kwargs
Pass extra parameters to `render()`:
```python theme={null}
def render(self, **kwargs) -> dict[str, str]:
format_type = kwargs.get("format", "json")
if format_type == "markdown":
docs = self._format_markdown()
elif format_type == "xml":
docs = self._format_xml()
else:
docs = json.dumps([d.model_dump() for d in self.documents])
return {"query": self.query, "documents": docs}
# Usage
session = PromptSession.from_prompt_template(
prompt=prompt,
session_data=input_data,
render_kwargs={"format": "markdown"}
)
```
## Schema Metadata
Generated models include metadata linking them to their source:
```python theme={null}
@classmethod
@property
def moxn_schema_metadata(cls) -> MoxnSchemaMetadata:
return MoxnSchemaMetadata(
schema_id="550e8400-e29b-41d4-a716-446655440000",
schema_version_id="version-uuid-if-from-commit",
prompt_id="prompt-uuid",
prompt_version_id="prompt-version-uuid",
task_id="task-uuid",
branch_id="branch-uuid-if-from-branch",
commit_id="commit-id-if-from-commit"
)
```
This enables:
* Creating sessions directly from session data
* Tracking which schema version was used in telemetry
* Validating that session data matches the expected prompt
## When to Regenerate
Regenerate models when:
* You add or modify variables in your prompts
* You change property types
* You add new prompts to your task
* You want to capture a new commit version
### Workflow suggestion
Run codegen as part of your development workflow:
```bash theme={null}
# In your project's Makefile or scripts
generate-models:
python -c "
import asyncio
from moxn import MoxnClient
async def main():
async with MoxnClient() as client:
await client.generate_task_models(
task_id='your-task-id',
branch_name='main',
output_dir='./src/generated'
)
asyncio.run(main())
"
```
Or add to CI/CD:
```yaml theme={null}
# GitHub Actions example
- name: Generate Moxn models
run: |
python scripts/generate_models.py
git diff --exit-code src/generated/
```
## Without Code Generation
You don't have to use codegen. You can create your own models:
```python theme={null}
from pydantic import BaseModel
from moxn.types.base import RenderableModel
class MyInput(RenderableModel):
query: str
context: list[str]
def render(self, **kwargs) -> dict[str, str]:
return {
"query": self.query,
"context": "\n".join(self.context)
}
# Use directly
session = await client.create_prompt_session(
prompt_id="...",
session_data=MyInput(query="...", context=["..."])
)
```
The key requirement is implementing the `RenderableModel` protocol:
```python theme={null}
class RenderableModel(Protocol):
def render(self, **kwargs) -> dict[str, str]:
"""Return flat string dict for variable substitution."""
...
@classmethod
@property
def moxn_schema_metadata(cls) -> MoxnSchemaMetadata | None:
"""Optional metadata linking to Moxn schema."""
...
```
## Type Mappings
JSON Schema types map to Python types:
| JSON Schema | Python Type |
| ------------------------------ | ----------------------- |
| `string` | `str` |
| `integer` | `int` |
| `number` | `float` |
| `boolean` | `bool` |
| `array` | `list[T]` |
| `object` | nested model or `dict` |
| `string` + `format: date` | `date` |
| `string` + `format: date-time` | `datetime` |
| `string` + `format: email` | `str` (with validation) |
| `string` + `format: uri` | `str` (with validation) |
## Next Steps
Use generated models with sessions
Understand how schemas work
See codegen in a complete example
Define variables in the web app
# Working with Prompts
Source: https://moxn.mintlify.app/guides/prompts
Fetching and managing prompts with MoxnClient
This guide covers how to fetch prompts and tasks from Moxn, including caching behavior and access patterns.
## MoxnClient
The `MoxnClient` is your primary interface to the Moxn API. Always use it as an async context manager:
```python theme={null}
from moxn import MoxnClient
async with MoxnClient() as client:
# Your code here
prompt = await client.get_prompt(...)
```
The context manager:
* Creates and manages HTTP connections
* Starts the telemetry dispatcher
* Ensures proper cleanup on exit
* Flushes pending telemetry events
### Configuration
The client reads configuration from environment variables:
| Variable | Description | Default |
| --------------- | ------------------------- | ---------------------- |
| `MOXN_API_KEY` | Your API key (required) | - |
| `MOXN_BASE_URL` | API base URL | `https://api.moxn.dev` |
| `MOXN_TIMEOUT` | Request timeout (seconds) | 30 |
## Fetching Prompts
### get\_prompt()
Fetch a single prompt by ID:
```python theme={null}
# By branch (development)
prompt = await client.get_prompt(
prompt_id="550e8400-e29b-41d4-a716-446655440000",
branch_name="main"
)
# By commit (production)
prompt = await client.get_prompt(
prompt_id="550e8400-e29b-41d4-a716-446655440000",
commit_id="abc123def456"
)
```
**Parameters:**
* `prompt_id`: UUID or string ID of the prompt
* `branch_name`: Branch to fetch from (mutually exclusive with `commit_id`)
* `commit_id`: Specific commit to fetch (mutually exclusive with `branch_name`)
**Returns:** `PromptTemplate` object
### get\_task()
Fetch an entire task with all its prompts:
```python theme={null}
task = await client.get_task(
task_id="task-uuid",
branch_name="main"
)
# Access prompts
for prompt in task.prompts:
print(f"{prompt.name}: {len(prompt.messages)} messages")
# Access schema definitions
for name, schema in task.definitions.items():
print(f"Schema: {name}")
```
**Parameters:**
* `task_id`: UUID or string ID of the task
* `branch_name`: Branch to fetch from
* `commit_id`: Specific commit to fetch
**Returns:** `Task` object containing all prompts and schemas
## Access Patterns
### Branch Access (Development)
Use `branch_name` during development to always get the latest version:
```python theme={null}
# Always fetches fresh data from the API
prompt = await client.get_prompt(
prompt_id="...",
branch_name="main"
)
```
**Characteristics:**
* Returns the latest state including uncommitted changes
* Not cached (always fetches fresh)
* May change between calls
* Includes working state modifications
### Commit Access (Production)
Use `commit_id` in production for immutable, reproducible prompts:
```python theme={null}
# Cached after first fetch
prompt = await client.get_prompt(
prompt_id="...",
commit_id="abc123def456"
)
```
**Characteristics:**
* Returns an immutable snapshot
* Cached indefinitely (commits never change)
* Guaranteed reproducibility
* No working state, only committed data
### Getting the Latest Commit
To get the latest committed state (without uncommitted changes), use a two-step pattern:
```python theme={null}
# Step 1: Get the branch head commit
head = await client.get_branch_head(
task_id="task-uuid",
branch_name="main"
)
# Step 2: Fetch by commit ID
prompt = await client.get_prompt(
prompt_id="...",
commit_id=head.effectiveCommitId
)
```
This is useful when you want the "stable" state of a branch without any work-in-progress changes.
## Caching Behavior
The SDK uses an in-memory cache for commit-based access:
```python theme={null}
# First call: fetches from API, stores in cache
prompt1 = await client.get_prompt("...", commit_id="abc123")
# Second call: returns from cache (no API call)
prompt2 = await client.get_prompt("...", commit_id="abc123")
# Branch access: always fetches (never cached)
prompt3 = await client.get_prompt("...", branch_name="main")
```
**Caching rules:**
* Commit-based fetches are cached (commits are immutable)
* Branch-based fetches always go to the API (branches change)
* Cache is per-client instance
* Cache is in-memory only (not persisted)
## PromptTemplate Structure
When you fetch a prompt, you get a `PromptTemplate` object:
```python theme={null}
prompt = await client.get_prompt(...)
# Core properties
prompt.id # UUID
prompt.name # str
prompt.description # str | None
prompt.task_id # UUID
# Content
prompt.messages # list[Message]
prompt.input_schema # Schema | None
# Versioning context
prompt.branch_id # UUID | None
prompt.commit_id # UUID | None
# Tools and completion config
prompt.tools # list[SdkTool]
prompt.completion_config # CompletionConfig | None
prompt.function_tools # list[SdkTool] (tool_type='tool')
prompt.structured_output_schema # SdkTool | None (tool_type='structured_output')
```
### Accessing Messages
```python theme={null}
# Get all messages
for message in prompt.messages:
print(f"[{message.role}] {message.name}")
for block_group in message.blocks:
for block in block_group:
print(f" - {block.block_type}")
# Get messages by role
system_messages = [m for m in prompt.messages if m.role == "system"]
user_messages = [m for m in prompt.messages if m.role == "user"]
# Get a specific message by role
system = prompt.get_message_by_role("system")
```
### Messages contain blocks
Messages use a 2D array structure for content blocks:
```python theme={null}
message.blocks # list[list[ContentBlock]]
# Outer list: paragraphs/sections
# Inner list: blocks within a paragraph
```
Block types include:
* `TextContent`: Plain text
* `Variable`: Template variables
* `ImageContentFromSource`: Images
* `PDFContentFromSource`: PDF documents
* `ToolCall` / `ToolResult`: Function calling
## Error Handling
Handle common errors when fetching prompts:
```python theme={null}
import httpx
async with MoxnClient() as client:
try:
prompt = await client.get_prompt(
prompt_id="...",
branch_name="main"
)
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
print("Prompt not found")
elif e.response.status_code == 403:
print("Not authorized to access this prompt")
else:
raise
```
## Verifying Access
Before making calls, you can verify your API key is valid:
```python theme={null}
async with MoxnClient() as client:
info = await client.verify_access()
print(f"Authenticated: {info['authenticated']}")
print(f"Tenant: {info['tenant_id']}")
```
## Next Steps
Combine prompts with runtime data
Convert prompts to provider formats
Understand branches and commits
Learn about Tasks, Prompts, and Messages
# Provider Integration
Source: https://moxn.mintlify.app/guides/providers
Converting prompts to Anthropic, OpenAI, and Google formats
Moxn supports multiple LLM providers. This guide covers how to convert prompts to provider-specific formats and handle their responses.
## Supported Providers
```python theme={null}
from moxn.types.content import Provider
Provider.ANTHROPIC # Claude (Anthropic)
Provider.OPENAI_CHAT # GPT models (OpenAI Chat Completions)
Provider.OPENAI_RESPONSES # OpenAI Responses API
Provider.GOOGLE_GEMINI # Gemini (Google AI Studio)
Provider.GOOGLE_VERTEX # Gemini (Vertex AI)
```
## Anthropic (Claude)
### Basic usage
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="...",
session_data=your_input
)
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# Log telemetry
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
### What to\_anthropic\_invocation() returns
```python theme={null}
{
"system": "Your system message...", # Optional
"messages": [
{"role": "user", "content": "..."},
{"role": "assistant", "content": "..."}
],
"model": "claude-sonnet-4-20250514", # From completion_config
"max_tokens": 4096,
"temperature": 0.7,
# If tools configured:
"tools": [...],
"tool_choice": {"type": "auto"},
# If structured output configured:
"output_format": {...}
}
```
### Extended thinking
For Claude models with extended thinking:
```python theme={null}
response = anthropic.messages.create(
**session.to_anthropic_invocation(
thinking={"type": "enabled", "budget_tokens": 10000}
),
extra_headers={"anthropic-beta": "interleaved-thinking-2025-05-14"}
)
```
### Structured outputs
If your prompt has a structured output schema configured:
```python theme={null}
response = anthropic.messages.create(
**session.to_anthropic_invocation(),
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
)
```
## OpenAI (GPT)
### Chat Completions API
```python theme={null}
from openai import OpenAI
openai = OpenAI()
response = openai.chat.completions.create(
**session.to_openai_chat_invocation()
)
# Log telemetry
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.OPENAI_CHAT
)
```
### What to\_openai\_chat\_invocation() returns
```python theme={null}
{
"messages": [
{"role": "system", "content": "..."},
{"role": "user", "content": "..."},
{"role": "assistant", "content": "..."}
],
"model": "gpt-4o",
"max_tokens": 4096,
"temperature": 0.7,
# If tools configured:
"tools": [...],
"tool_choice": "auto",
"parallel_tool_calls": True,
# If structured output configured:
"response_format": {...}
}
```
### Responses API
For OpenAI's newer Responses API:
```python theme={null}
response = openai.responses.create(
**session.to_openai_responses_invocation()
)
await client.log_telemetry_event_from_response(
session, response, Provider.OPENAI_RESPONSES
)
```
### Reasoning models
For o1, o3, and other reasoning models:
```python theme={null}
response = openai.chat.completions.create(
**session.to_openai_chat_invocation(
thinking={"reasoning_effort": "high"}
)
)
```
## Google (Gemini)
### Google AI Studio
```python theme={null}
from google import genai
google_client = genai.Client()
response = google_client.models.generate_content(
**session.to_google_gemini_invocation()
)
# Log telemetry
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.GOOGLE_GEMINI
)
```
### Vertex AI
```python theme={null}
from google import genai
vertex_client = genai.Client(vertexai=True)
response = vertex_client.models.generate_content(
**session.to_google_vertex_invocation()
)
await client.log_telemetry_event_from_response(
session, response, Provider.GOOGLE_VERTEX
)
```
### What to\_google\_gemini\_invocation() returns
```python theme={null}
{
"model": "gemini-2.5-flash",
"contents": [...], # Conversation content
"config": {
"system_instruction": "...",
"max_output_tokens": 4096,
"temperature": 0.7,
# If tools configured:
"tools": [{"function_declarations": [...]}],
"tool_config": {...},
# If structured output:
"response_schema": {...},
"response_mime_type": "application/json"
}
}
```
### Thinking models
For Gemini thinking models:
```python theme={null}
response = google_client.models.generate_content(
**session.to_google_gemini_invocation(
thinking={"thinking_budget": 10000}
)
)
```
## Generic Provider Method
Use `to_invocation()` for provider-agnostic code:
```python theme={null}
from moxn.types.content import Provider
# Use provider from prompt's completion_config
payload = session.to_invocation()
# Or specify explicitly
payload = session.to_invocation(provider=Provider.ANTHROPIC)
# With overrides
payload = session.to_invocation(
provider=Provider.OPENAI_CHAT,
model="gpt-4o",
max_tokens=8000,
temperature=0.5
)
```
## Message-Only Methods
If you only need messages (without model config):
```python theme={null}
# Generic method (uses stored provider from completion_config)
messages = session.to_messages()
# Or override provider explicitly
messages = session.to_messages(provider=Provider.ANTHROPIC)
# Or use provider-specific methods
anthropic_payload = session.to_anthropic_messages() # {system, messages}
openai_payload = session.to_openai_chat_messages() # {messages}
google_payload = session.to_google_gemini_messages() # {system_instruction, content}
```
## Parsing Responses
Parse any provider's response to a normalized format:
```python theme={null}
# Parse response (uses stored provider from completion_config)
parsed = session.parse_response(response)
# Or override provider explicitly
# parsed = session.parse_response(response, provider=Provider.ANTHROPIC)
# Access normalized data
parsed.candidates # list[Candidate] - response options
parsed.input_tokens # int | None
parsed.output_tokens # int | None
parsed.model # str | None
parsed.stop_reason # str | None
parsed.raw_response # dict - original response
parsed.provider # Provider
# Each candidate has content blocks
for candidate in parsed.candidates:
for block in candidate.content:
match block.block_type:
case "text":
print(block.text)
case "tool_call":
print(f"{block.tool_name}: {block.input}")
case "thinking":
print(f"Thinking: {block.text}")
```
## Tool Use
If your prompt has tools configured, they're automatically included:
```python theme={null}
# Tools are included in the invocation
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# Check for tool calls in response
parsed = session.parse_response(response)
for candidate in parsed.candidates:
for block in candidate.content:
if block.block_type == "tool_call":
# Execute the tool
result = execute_tool(block.tool_name, block.input)
# Add tool result to session (if doing multi-turn)
# Then call the LLM again...
```
### Tool choice
The SDK translates tool\_choice across providers:
| Moxn Setting | Anthropic | OpenAI | Google |
| ------------- | --------------------------------- | --------------------------------------------------- | ---------------------- |
| `"auto"` | `{"type": "auto"}` | `"auto"` | `{"mode": "AUTO"}` |
| `"required"` | `{"type": "any"}` | `"required"` | `{"mode": "ANY"}` |
| `"none"` | Tools omitted | `"none"` | `{"mode": "NONE"}` |
| `"tool_name"` | `{"type": "tool", "name": "..."}` | `{"type": "function", "function": {"name": "..."}}` | `{"mode": "ANY", ...}` |
## Multimodal Content
Images and PDFs are automatically converted to provider-specific formats:
```python theme={null}
# Anthropic: base64 with media_type
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": "..."
}
}
# OpenAI: data URI
{
"type": "image_url",
"image_url": {"url": "data:image/png;base64,..."}
}
# Google: Part with inline_data
Part(inline_data=Blob(mime_type="image/png", data=...))
```
The SDK handles signed URL refresh automatically for images and PDFs stored in cloud storage.
## Error Handling
Handle provider-specific errors:
```python theme={null}
from anthropic import APIError as AnthropicError
from openai import APIError as OpenAIError
try:
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
except AnthropicError as e:
if "rate_limit" in str(e):
# Handle rate limiting
await asyncio.sleep(60)
else:
raise
```
## Provider Feature Matrix
| Feature | Anthropic | OpenAI Chat | OpenAI Responses | Google |
| ----------------- | -------------- | ----------- | ---------------- | --------- |
| System messages | Separate field | In messages | Instructions | Config |
| Tools | Yes | Yes | Yes | Yes |
| Structured output | Yes (beta) | Yes | Yes | Yes |
| Images | Yes | Yes | Yes | Yes |
| PDFs | Yes | Yes | Limited | Yes |
| Extended thinking | Yes | Yes (o1/o3) | Yes | Yes |
| Streaming | SDK level | SDK level | SDK level | SDK level |
## Next Steps
Generate type-safe input models
Configure and parse structured responses
Work with images and documents
Log provider interactions
# Prompt Sessions
Source: https://moxn.mintlify.app/guides/sessions
Runtime prompt execution with PromptSession
A `PromptSession` is the runtime representation of a prompt combined with your input data. It handles variable substitution, provider conversion, and response parsing.
## Creating Sessions
### From prompt ID
The simplest way to create a session:
```python theme={null}
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
branch_name="main",
session_data=YourInputModel(
query="How do I reset my password?",
user_id="user_123"
)
)
```
This fetches the prompt and creates a session in one call.
### From session data with metadata
If your codegen'd model has `moxn_schema_metadata`, you can create a session from just the data:
```python theme={null}
from generated_models import ProductHelpInput
session_data = ProductHelpInput(
query="How do I reset my password?",
user_id="user_123"
)
# Session data knows its prompt ID from metadata
session = await client.prompt_session_from_session_data(
session_data=session_data,
branch_name="main"
)
```
### From a PromptTemplate directly
For more control, create sessions manually:
```python theme={null}
from moxn.models.prompt import PromptSession
prompt = await client.get_prompt("...", branch_name="main")
session = PromptSession.from_prompt_template(
prompt=prompt,
session_data=YourInputModel(...)
)
```
## Session Data and Rendering
### The render() flow
Session data goes through a transformation pipeline:
```mermaid theme={null}
flowchart TB
A["YourInputModel (Pydantic)"] --> B[".render()"]
B --> C["dict[str, RenderValue]"]
C --> D["Variable substitution in messages"]
D --> E["Provider-specific format"]
subgraph RenderValue["RenderValue Types"]
S["str - text values"]
CB["ContentBlock - images, PDFs"]
ARR["Sequence[ContentBlock] - arrays"]
end
```
The `render()` method returns a dictionary where values can be:
* **Strings**: Simple text values (the base case)
* **ContentBlocks**: Multimodal content like images and PDFs
* **Arrays**: Sequences of ContentBlocks for structured context (e.g., citations)
### How render() works
The `render()` method transforms your typed data into a dictionary that maps variable names to values. The base case returns string values:
```python theme={null}
class ProductHelpInput(RenderableModel):
query: str
user_id: str
documents: list[Document]
def render(self, **kwargs) -> dict[str, str]:
return {
"query": self.query,
"user_id": self.user_id,
"documents": json.dumps([d.model_dump() for d in self.documents])
}
```
Each key in the returned dict corresponds to a variable name in your prompt messages.
For multimodal content and structured arrays, `render()` can return ContentBlock types instead of strings—see [Advanced Render Patterns](#advanced-render-patterns) below.
### Customizing render()
You can customize how data is formatted for the LLM:
```python theme={null}
class ProductHelpInput(RenderableModel):
documents: list[Document]
def render(self, **kwargs) -> dict[str, str]:
# Format as markdown
formatted = "\n".join([
f"## {doc.title}\n{doc.content}"
for doc in self.documents
])
return {
"documents": formatted
}
```
Or use XML formatting:
```python theme={null}
def render(self, **kwargs) -> dict[str, str]:
docs_xml = "\n".join([
f"\n{doc.content}\n"
for doc in self.documents
])
return {
"documents": f"\n{docs_xml}\n"
}
```
### Advanced Render Patterns
Beyond simple strings, `render()` can return ContentBlock types for multimodal content and arrays for structured context.
#### Multimodal Variables
Inject images and PDFs directly into your prompts:
```python theme={null}
from moxn.base_models.blocks.image import ImageContentFromSource
from moxn.base_models.blocks.file import PDFContentFromSource
class DocumentAnalysisInput(RenderableModel):
query: str
document_url: str
screenshot_url: str
def render(self, **kwargs) -> dict[str, Any]:
return {
"query": self.query,
"document": PDFContentFromSource(
url=self.document_url,
name="source_document.pdf",
),
"screenshot": ImageContentFromSource(
url=self.screenshot_url,
),
}
```
When the session converts to provider format, these ContentBlocks become the appropriate multimodal content (e.g., `image` and `document` blocks for Anthropic).
#### Arrays for Citations
Return arrays of ContentBlocks to enable structured citation support. This is particularly useful with Anthropic's citations API, which allows the model to explicitly reference specific chunks of context in its response:
```python theme={null}
from moxn.base_models.blocks.text import TextContent
class RAGInput(RenderableModel):
query: str
chunks: list[DocumentChunk]
def render(self, **kwargs) -> dict[str, Any]:
return {
"query": self.query,
# Array of TextContent blocks for citations
"context_chunks": [
TextContent(text=chunk.text)
for chunk in self.chunks
],
}
```
When Anthropic's citations feature is enabled, each `TextContent` in the array becomes a citable source. The model's response can then reference specific chunks by index, making it easy to verify which sources informed each part of the response.
Arrays of ContentBlocks work with any provider, but citation-aware responses are currently specific to Anthropic's citations API.
## Converting to Provider Format
Sessions convert to provider-specific formats using the `to_*_invocation()` methods:
### Anthropic
```python theme={null}
from anthropic import Anthropic
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
```
The invocation includes:
* `system`: System message (if present)
* `messages`: User/assistant messages
* `model`: From prompt's completion\_config
* `max_tokens`: From completion\_config
* `tools`: If tools are configured
* `output_format`: If structured output is configured
### OpenAI
```python theme={null}
from openai import OpenAI
openai = OpenAI()
response = openai.chat.completions.create(
**session.to_openai_chat_invocation()
)
```
### Google
```python theme={null}
from google import genai
client = genai.Client()
response = client.models.generate_content(
**session.to_google_gemini_invocation()
)
```
### Generic method
Use `to_invocation()` which automatically selects the provider:
```python theme={null}
from moxn.types.content import Provider
# Uses completion_config.provider if set
payload = session.to_invocation()
# Or specify explicitly
payload = session.to_invocation(provider=Provider.ANTHROPIC)
```
### Provider Defaulting
When your prompt has `completion_config.provider` set (configured in the Moxn web app), many methods use it automatically:
```python theme={null}
# These all use the stored provider from completion_config:
payload = session.to_invocation() # Invocation payload
payload = session.to_payload() # Messages-only payload
parsed = session.parse_response(response) # Response parsing
event = session.create_llm_event_from_response(response) # Telemetry
# Override when needed:
parsed = session.parse_response(response, provider=Provider.OPENAI_CHAT)
```
This simplifies code when you're using the provider configured in the web app, while still allowing runtime overrides.
### Overriding parameters
Override model parameters at runtime:
```python theme={null}
response = anthropic.messages.create(
**session.to_anthropic_invocation(
model="claude-sonnet-4-20250514", # Override model
max_tokens=4096, # Override max tokens
temperature=0.7 # Override temperature
)
)
```
## Adding Runtime Messages
Append messages to a session after creation:
### Add user message
```python theme={null}
session.append_user_text(
text="Here's additional context...",
name="Additional Context"
)
```
### Add assistant message
```python theme={null}
session.append_assistant_text(
text="I understand. Let me help with that.",
name="Assistant Acknowledgment"
)
```
### Add from LLM response
After getting a response, add it to the session for multi-turn conversations:
```python theme={null}
# Parse the response (uses stored provider from completion_config)
parsed = session.parse_response(response)
# Add to session
session.append_assistant_response(
parsed_response=parsed,
name="Assistant Response"
)
# Now add a follow-up user message
session.append_user_text("Can you elaborate on that?")
# Send again
response2 = anthropic.messages.create(
**session.to_anthropic_invocation()
)
```
## Response Parsing
Parse provider responses into a normalized format:
```python theme={null}
# Get raw response from provider
response = anthropic.messages.create(...)
# Parse to normalized format (uses stored provider from completion_config)
parsed = session.parse_response(response)
# Or override the provider explicitly
# parsed = session.parse_response(response, provider=Provider.ANTHROPIC)
# Access normalized content
for candidate in parsed.candidates:
for block in candidate.content:
if block.block_type == "text":
print(block.text)
elif block.block_type == "tool_call":
print(f"Tool: {block.tool_name}({block.input})")
# Access metadata
print(f"Tokens: {parsed.input_tokens} in, {parsed.output_tokens} out")
print(f"Model: {parsed.model}")
print(f"Stop reason: {parsed.stop_reason}")
```
## Creating Telemetry Events
Create LLM events from responses for logging:
```python theme={null}
# Method 1: From raw response (uses stored provider from completion_config)
llm_event = session.create_llm_event_from_response(response)
# Or override the provider explicitly
# llm_event = session.create_llm_event_from_response(response, provider=Provider.ANTHROPIC)
# Method 2: From parsed response (with more control)
parsed = session.parse_response(response)
llm_event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
request_config=request_config, # Optional
schema_definition=schema, # Optional
attributes={"custom": "data"}, # Optional
validation_errors=errors # Optional
)
# Log it
async with client.span(session) as span:
await client.log_telemetry_event(llm_event)
```
## Session Properties
Access session information:
```python theme={null}
session.id # UUID - unique session identifier
session.prompt_id # UUID - source prompt ID
session.prompt # PromptTemplate - the underlying prompt
session.messages # list[Message] - current messages (with appended)
session.session_data # RenderableModel | None - your input data
```
## Complete Example
Here's a full multi-turn conversation example:
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
from generated_models import ChatInput
async def chat_conversation():
async with MoxnClient() as client:
# Create initial session
session = await client.create_prompt_session(
prompt_id="chat-prompt-id",
branch_name="main",
session_data=ChatInput(
user_name="Alice",
system_context="You are a helpful assistant."
)
)
anthropic = Anthropic()
# First turn
session.append_user_text("What's the weather like today?")
async with client.span(session, name="turn_1") as span:
response1 = anthropic.messages.create(
**session.to_anthropic_invocation()
)
await client.log_telemetry_event_from_response(
session, response1, Provider.ANTHROPIC
)
# Add response to session
parsed1 = session.parse_response(response1)
session.append_assistant_response(parsed1)
# Second turn
session.append_user_text("What about tomorrow?")
async with client.span(session, name="turn_2") as span:
response2 = anthropic.messages.create(
**session.to_anthropic_invocation()
)
await client.log_telemetry_event_from_response(
session, response2, Provider.ANTHROPIC
)
return response2.content[0].text
```
## Next Steps
Deep dive into provider-specific handling
Generate type-safe session data models
Set up observability for sessions
Understand how variables work
# The Moxn Workflow
Source: https://moxn.mintlify.app/guides/workflow
The end-to-end workflow for building with Moxn
This guide covers the complete development workflow with Moxn—from creating prompts in the web app to deploying type-safe LLM applications with full observability.
## The Development Cycle
```mermaid theme={null}
flowchart LR
A[CREATE: Prompts in Web App] --> B[GENERATE: Type-safe Models]
B --> C[USE: Prompts in Your Code]
C --> D[LOG: Telemetry to Moxn]
```
## Step 1: Create Prompts in the Web App
### Create a Task
A **Task** is a container for related prompts—think of it like a Git repository. Create one for each distinct AI feature in your application.
![Tasks list]()
Tasks contain related prompts, schemas, and traces:
* **Customer Support Bot** (Task)
* Product Help (Prompt)
* Query Classification (Prompt)
* Escalation Handler (Prompt)
![Task detail]()
### Add Messages and Variables
Each prompt contains **messages** with roles (system, user, assistant). Insert **variables** using the `/variable` slash command in the editor, which opens a property editor.
![Message editor with variables]()
Variables are typed properties—not template strings. When you insert a variable:
1. Type `/variable` in the message editor
2. The **Property Editor** opens where you configure the variable
3. Set the name, type (string, array, object, image-url, etc.), and optional schema reference
4. The variable appears as a styled block in your message
![Property editor]()
Variables automatically sync to the prompt's **Input Schema**—the typed interface your code will use:
![Prompt input schema]()
### Commit Your Changes
When you're happy with your prompts, commit them. This creates an immutable snapshot you can reference by commit ID in production.
## Step 2: Generate Type-Safe Models
Generate Pydantic models from your prompt schemas:
```python theme={null}
import asyncio
from moxn import MoxnClient
async def generate():
async with MoxnClient() as client:
await client.generate_task_models(
task_id="your-task-id",
branch_name="main",
output_dir="./generated_models"
)
asyncio.run(generate())
```
This creates a Python file with models for each prompt:
```python theme={null}
# generated_models/customer_support_bot_models.py
class ProductHelpInput(RenderableModel):
"""Input schema for Product Help prompt."""
company_name: str
customer_name: str
query: str
search_results: list[SearchResult]
def render(self, **kwargs) -> dict[str, str]:
return {
"company_name": self.company_name,
"customer_name": self.customer_name,
"query": self.query,
"search_results": json.dumps([r.model_dump() for r in self.search_results]),
}
```
**Codegen is optional.** You can always define models manually—especially useful when iterating on prompts in a notebook or during early development. Codegen shines when you want to ensure your models stay in sync with your prompt schemas, and integrates well into CI/CD pipelines (similar to database migrations or OpenAPI client generation).
You don't need generated models to use Moxn. Define your input model manually:
```python theme={null}
from moxn.types.base import RenderableModel
class MyInput(RenderableModel):
query: str
context: str
def render(self, **kwargs) -> dict[str, str]:
return {"query": self.query, "context": self.context}
# Use it directly
session = await client.create_prompt_session(
prompt_id="...",
session_data=MyInput(query="Hello", context="Some context")
)
```
This is often the fastest way to iterate when you're actively editing prompts and experimenting with different variable structures.
## Step 3: Use Prompts in Your Code
Now use the generated models with the SDK:
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from generated_models.customer_support_bot_models import ProductHelpInput
from anthropic import Anthropic
async def handle_support_query(customer_name: str, query: str):
async with MoxnClient() as client:
# Create session with type-safe input
session = await client.create_prompt_session(
prompt_id="product-help-prompt-id",
branch_name="main", # Use commit_id in production
session_data=ProductHelpInput(
company_name="Acme Corp",
customer_name=customer_name,
query=query,
search_results=await search_knowledge_base(query)
)
)
# Send to LLM
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
return response.content[0].text
```
### Branch vs Commit Access
| Pattern | When to Use | Example |
| -------------------- | -------------------- | ------------------ |
| `branch_name="main"` | Development, testing | Always gets latest |
| `commit_id="abc123"` | Production | Immutable, pinned |
```python theme={null}
# Development: always get latest
session = await client.create_prompt_session(
prompt_id="...",
branch_name="main"
)
# Production: pin to specific commit
session = await client.create_prompt_session(
prompt_id="...",
commit_id="abc123def456"
)
```
## Step 4: Log Telemetry
Log every LLM interaction for debugging, analysis, and compliance:
```python theme={null}
async def handle_support_query(customer_name: str, query: str):
async with MoxnClient() as client:
session = await client.create_prompt_session(...)
# Wrap your LLM call in a span with searchable metadata
async with client.span(
session,
name="support_response",
metadata={
"customer_name": customer_name,
"query_type": classify_query(query)
}
) as span:
# Make the LLM call
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# Log the event
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
return response.content[0].text
```
### What Gets Logged
Each telemetry event captures:
* **Session data**: The typed input you provided
* **Rendered input**: The flattened string values injected into the prompt
* **Messages**: The complete prompt sent to the LLM
* **Response**: The raw LLM response
* **Metadata**: Span attributes, token counts, latency
## Viewing Results
After logging telemetry, view your traces in the Moxn web app:
1. Navigate to your Task
2. Open the **Traces** tab
3. Filter by time range, prompt, or custom attributes
4. Click a trace to see the full span hierarchy
![Traces list]()
You'll see:
* Complete input/output for each LLM call
* Token usage and costs
* Latency breakdowns
* Custom attributes for filtering
## Iterating on Prompts
The workflow creates a tight feedback loop:
1. **Observe**: Review traces to see how prompts perform in production
2. **Branch**: Create a branch for experimentation
3. **Edit**: Modify prompts in the web app
4. **Test**: Use branch access to test changes
5. **Commit**: When satisfied, commit and update production
```python theme={null}
# Test a branch
session = await client.create_prompt_session(
prompt_id="...",
branch_name="experiment-new-tone" # Your feature branch
)
# Deploy to production
session = await client.create_prompt_session(
prompt_id="...",
commit_id="new-commit-id" # After committing the branch
)
```
## Next Steps
Learn the details of fetching and caching prompts
Understand sessions and variable rendering
Convert to Anthropic, OpenAI, and Google formats
Deep dive into model generation
# Moxn
Source: https://moxn.mintlify.app/index
Version, render, and observe your LLM prompts
```mermaid theme={null}
flowchart LR
A[Author] --> B[Generate]
B --> C[Use]
C --> D[Observe]
```
Build prompts with typed variables, structured content, and Git-like
versioning.
Run codegen to get Pydantic models with autocomplete and validation for your
LLM invocations.
Fetch prompts, inject your context, get provider specific dialects prepared,
no more shoe horning.
Log traces and review them in the same editor you authored in.
## Quick Example
```python theme={null}
from moxn import MoxnClient
from anthropic import Anthropic
async with MoxnClient() as client:
# Fetch prompt and inject your data
session = await client.create_prompt_session(
prompt_id="support-agent",
session_data=SupportInput(query="How do I reset my password?")
)
# Send to provider
response = Anthropic().messages.create(
**session.to_anthropic_invocation()
)
# Log for observability
await client.log_telemetry_event_from_response(session, response)
```
Install and run your first prompt
The problems we solve
**For AI tools:** This documentation is available as [llms.txt](/llms.txt) and [llms-full.txt](/llms-full.txt) for LLM consumption.
# llms.txt
Source: https://moxn.mintlify.app/llms-txt
Machine-readable documentation for AI tools
This documentation is optimized for AI consumption through industry-standard llms.txt files.
## Available Files
Index of all documentation pages with descriptions
Complete documentation content in a single file
## What is llms.txt?
The `llms.txt` file is an industry standard that helps LLMs index documentation content more efficiently. It contains:
* Site title and description
* A structured list of all pages with their descriptions
* Links to each page in the documentation
The `llms-full.txt` file combines the entire documentation site into a single file, making it easy for AI tools to ingest all content at once.
## Usage
You can access these files by appending `/llms.txt` or `/llms-full.txt` to the documentation URL. AI assistants like Claude, ChatGPT, and Cursor can use these files to better understand and reference the Moxn SDK documentation.
# Quick Start
Source: https://moxn.mintlify.app/quickstart
Get up and running with Moxn in 5 minutes
This guide walks you through installing the SDK, fetching a prompt, and making your first LLM call with telemetry.
## Prerequisites
* Python 3.10+
* A Moxn account and API key (get one at [moxn.dev](https://moxn.dev))
* An LLM provider API key (Anthropic, OpenAI, or Google)
## Installation
Install the Moxn SDK using pip:
```bash theme={null}
pip install moxn
```
Set your API key as an environment variable:
```bash theme={null}
export MOXN_API_KEY="your-api-key"
```
## Your First Moxn Call
Here's a complete example that fetches a prompt, creates a session, sends it to Claude, and logs the interaction:
```python Anthropic theme={null}
import asyncio
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
async def main():
# Initialize the Moxn client
async with MoxnClient() as client:
# Fetch a prompt from your task
# Use branch_name for development, commit_id for production
prompt = await client.get_prompt(
prompt_id="your-prompt-id",
branch_name="main"
)
# Create a session with your input data
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
branch_name="main",
session_data=YourPromptInput(
ragItems=[
ragItem(
title="Best Document",
content="The answer to the user's query"
)
]
)
)
# Convert to Anthropic format and send
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
print(response.content[0].text)
# Log telemetry
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
asyncio.run(main())
```
```python OpenAI theme={null}
import asyncio
from moxn import MoxnClient
from moxn.types.content import Provider
from openai import OpenAI
async def main():
# Initialize the Moxn client
async with MoxnClient() as client:
# Fetch a prompt from your task
prompt = await client.get_prompt(
prompt_id="your-prompt-id",
branch_name="main"
)
# Create a session with your input data
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
branch_name="main",
session_data=None # Replace with your codegen'd model
)
# Convert to OpenAI format and send
openai = OpenAI()
response = openai.chat.completions.create(
**session.to_openai_chat_invocation()
)
print(response.choices[0].message.content)
# Log telemetry
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.OPENAI_CHAT
)
asyncio.run(main())
```
```python Google theme={null}
import asyncio
from moxn import MoxnClient
from moxn.types.content import Provider
from google import genai
async def main():
# Initialize the Moxn client
async with MoxnClient() as client:
# Create a session with your input data
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
branch_name="main",
session_data=None # Replace with your codegen'd model
)
# Convert to Google format and send
google_client = genai.Client()
response = google_client.models.generate_content(
**session.to_google_gemini_invocation()
)
print(response.text)
# Log telemetry
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.GOOGLE_GEMINI
)
asyncio.run(main())
```
## From Template to Provider Call
The code above follows the core Moxn pattern (see [Core Concepts](/index#core-concepts)):
| Step | What Happens |
| ----------------------- | ---------------------------------------------------------------- |
| **1. Fetch template** | Your prompt (messages, variables, config) is retrieved from Moxn |
| **2. Create session** | Template + your `session_data` = a session ready to render |
| **3. Build invocation** | `to_anthropic_invocation()` returns a plain Python dict |
| **4. Call provider** | You pass the dict to the provider SDK with `**invocation` |
### You Control the Payload
The invocation is just a dictionary—you can inspect, modify, or extend it:
```python theme={null}
invocation = session.to_anthropic_invocation()
# Add streaming
invocation["stream"] = True
# Use a new provider feature
invocation["extra_headers"] = {"anthropic-beta": "citations-2025-01-01"}
# Override settings for A/B testing
invocation["model"] = "claude-sonnet-4-20250514"
response = anthropic.messages.create(**invocation)
```
This design is intentional: Moxn helps you **build payloads**, but never **owns the integration**. You always call the provider SDK directly—no wrapper, no middleware, no magic. If a provider releases a new feature tomorrow, you can use it immediately by adding parameters to the dict or to the provider directly.
Moxn can help you build the provider specific payload.
## Understanding the Code
Let's break down what's happening:
### 1. MoxnClient as Context Manager
```python theme={null}
async with MoxnClient() as client:
```
The client manages connections and telemetry batching. Always use it as an async context manager to ensure proper cleanup.
### 2. Fetching Prompts
```python theme={null}
prompt = await client.get_prompt(
prompt_id="your-prompt-id",
branch_name="main" # or commit_id="abc123" for production
)
```
* **Branch access** (`branch_name`): Gets the latest version, including uncommitted changes. Use for development.
* **Commit access** (`commit_id`): Gets an immutable snapshot. Use for production.
### 3. Creating Sessions
```python theme={null}
prompt_session = await client.create_prompt_session(
prompt_id="your-prompt-id",
branch_name="main",
session_data=YourInputModel(...) # Pydantic model from codegen
)
```
A (prompt) session combines your prompt template with runtime data. The `session_data` is typically a Pydantic model generated by codegen.
The (prompt) session holds the message history - you can append additional messages or context, append an LLM response followed by additional user messages.
It maintains the conversation history for the session. Each telemetry loggign call will log the session independently.
### 4. Converting to Provider Format
```python theme={null}
response = anthropic.messages.create(
**prompt_session.to_anthropic_invocation()
)
```
The `to_*_invocation()` methods return complete payloads you can unpack directly into provider SDKs. They include:
* Messages formatted for the provider
* Model configuration from your prompt
* Tools and structured output schemas (if configured)
### 5. Logging Telemetry
```python theme={null}
async with client.span(session) as span:
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
Spans create observable traces of your LLM calls. Every call within a span is linked for debugging and analysis.
## Using Code Generation
For type-safe session data, generate Pydantic models from your prompts:
```python theme={null}
# Generate models for all prompts in a task
async with MoxnClient() as client:
await client.generate_task_models(
task_id="your-task-id",
branch_name="main",
output_dir="./models"
)
```
This creates a Python file with Pydantic models matching your prompt's input schema:
```python theme={null}
# models/your_task_models.py (generated)
from datetime import datetime
from typing import TypedDict
from pydantic import BaseModel
from moxn.types.base import RenderableModel
class DocumentRendered(TypedDict):
"""Flattened string representation for prompt template injection."""
title: str
created_at: str
last_modified: str
author: str
content: str
class Document(BaseModel):
title: str
created_at: datetime
last_modified: datetime | None = None
author: str
content: str
def render(self, **kwargs) -> DocumentRendered:
"""Render to flattened dictionary for prompt variable substitution."""
result: DocumentRendered = {
"title": self.title,
"created_at": self.created_at.isoformat(),
"last_modified": self.last_modified.isoformat() if self.last_modified else "",
"author": self.author,
"content": self.content,
}
return result
class YourPromptInputRendered(TypedDict):
"""Flattened representation - keys match your prompt's variable blocks."""
query: str
user_id: str
context: str
class YourPromptInput(RenderableModel):
query: str
user_id: str
context: list[Document] | None = None
def render(self, **kwargs) -> YourPromptInputRendered:
# Render context as XML document collection
if self.context:
doc_xmls = []
for doc in self.context:
data = doc.render(**kwargs) # Returns DocumentRendered
attrs = f'title="{data["title"]}" author="{data["author"]}" created_at="{data["created_at"]}"'
if data["last_modified"]:
attrs += f' last_modified="{data["last_modified"]}"'
doc_xmls.append(f"\n{data['content']}\n")
context_str = "\n" + "\n".join(doc_xmls) + "\n"
else:
context_str = ""
return {
"query": self.query,
"user_id": self.user_id,
"context": context_str,
}
```
The `render()` method transforms your typed data into strings for prompt injection. This example renders documents as XML—a format that works well for providing structured context to LLMs.
Then use it in your code:
```python theme={null}
from models.your_task_models import YourPromptInput, Document
from datetime import datetime
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
session_data=YourPromptInput(
query="How do I reset my password?",
user_id="user_123",
context=[
Document(
title="Password Reset Guide",
created_at=datetime(2024, 1, 15),
author="Support Team",
content="To reset your password, click 'Forgot Password' on the login page..."
),
Document(
title="Account Security FAQ",
created_at=datetime(2024, 2, 1),
last_modified=datetime(2024, 3, 10),
author="Security Team",
content="We recommend using a password manager and enabling 2FA..."
)
]
)
)
```
This renders into the prompt as:
```xml theme={null}
To reset your password, click 'Forgot Password' on the login page...
We recommend using a password manager and enabling 2FA...
```
### The Render Pipeline
When you create a prompt session, your typed data transforms through several stages:
```
┌─────────────────────────────────────────────────────────────────────────┐
│ YOUR CODE │
│ ───────── │
│ YourPromptInput( ← Pydantic BaseModel │
│ query="How do I reset...", (typed, validated) │
│ user_id="user_123", │
│ context=[Document(...), ...] │
│ ) │
└─────────────────────────────────────────────────────────────────────────┘
│
│ .render(**kwargs)
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ RENDERED REPRESENTATION │
│ ─────────────────────── │
│ YourPromptInputRendered = { ← TypedDict (all string values) │
│ "query": "How do I reset...", │
│ "user_id": "user_123", │
│ "context": "...", │
│ } │
└─────────────────────────────────────────────────────────────────────────┘
│
│ to_anthropic_invocation()
│ (variables matched by name)
▼
┌─────────────────────────────────────────────────────────────────────────┐
│ PROVIDER PAYLOAD │
│ ──────────────── │
│ { │
│ "model": "claude-sonnet-4-20250514", │
│ "messages": [ │
│ {"role": "user", "content": "How do I reset... ..."} │
│ ], │
│ ... │
│ } │
└─────────────────────────────────────────────────────────────────────────┘
```
### How It Works
When you create a prompt session with `session_data`, the SDK:
1. **Stores your typed model** — The Pydantic instance you pass to `create_prompt_session()`
2. **Calls `render()` at invocation time** — When you call `to_anthropic_invocation()` (or other providers)
3. **Substitutes variables by name** — Each key in the rendered dict matches a `{{variable}}` block in your prompt
```python theme={null}
# Your code
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
session_data=YourPromptInput(...) # ← Stored as-is
)
# When you call this, render() is invoked automatically
invocation = session.to_anthropic_invocation()
# session_data.render() → YourPromptInputRendered → variable substitution
```
**Telemetry captures both representations**:
* **Session data**: Your original typed model (enables re-rendering with different logic)
* **Rendered input**: The flattened strings that were injected into the prompt
**Why Two Representations?**
The dual-representation pattern serves distinct purposes:
| Representation | Type | Purpose |
| ---------------------------- | ---------------- | -------------------------------------------------------------------------------------- |
| **BaseModel** (session data) | Pydantic model | Type-safe data structure with validation, rich types (datetime, nested objects, lists) |
| **\*Rendered** (TypedDict) | `dict[str, str]` | Flat string values ready for template injection |
This separation enables:
* **Re-rendering**: Change how data is formatted without changing the data itself
* **Telemetry**: Log the structured input separately from the rendered output
* **Testing**: Compare different rendering strategies using the same session data
## Next Steps
Understand the full development workflow
Learn more about sessions and rendering
Deep dive into type-safe model generation
Set up comprehensive observability
# MoxnClient
Source: https://moxn.mintlify.app/reference/client
API reference for the MoxnClient class
The `MoxnClient` is your main entry point for interacting with the Moxn platform. It handles authentication, fetching prompts, creating sessions, managing telemetry, and code generation.
## Usage
```python theme={null}
from moxn import MoxnClient
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="your-prompt-id",
branch_name="main"
)
```
The client automatically uses `MOXN_API_KEY` from your environment.
## Fetching Content
### get\_prompt
Fetch a prompt template with optional caching.
```python theme={null}
async def get_prompt(
prompt_id: UUID | str,
branch_name: str | None = None,
commit_id: str | None = None,
) -> PromptTemplate
```
**Parameters:**
| Name | Type | Description |
| ------------- | ------------- | --------------------------------------------------------------------- |
| `prompt_id` | `UUID \| str` | The prompt's anchor ID |
| `branch_name` | `str \| None` | Branch name (mutually exclusive with commit\_id) |
| `commit_id` | `str \| None` | Commit ID for immutable access (mutually exclusive with branch\_name) |
**Returns:** `PromptTemplate`
**Caching Behavior:**
* **Commit access**: Cached indefinitely (immutable)
* **Branch access**: Always fetches latest (mutable)
```python theme={null}
# Development: get latest from branch
prompt = await client.get_prompt("prompt-id", branch_name="main")
# Production: pin to specific commit
prompt = await client.get_prompt("prompt-id", commit_id="abc123")
```
### get\_task
Fetch an entire task with all its prompts and schemas.
```python theme={null}
async def get_task(
task_id: str,
branch_name: str | None = None,
commit_id: str | None = None,
) -> Task
```
**Parameters:**
| Name | Type | Description |
| ------------- | ------------- | -------------------- |
| `task_id` | `str` | The task's anchor ID |
| `branch_name` | `str \| None` | Branch name |
| `commit_id` | `str \| None` | Commit ID |
**Returns:** `Task`
```python theme={null}
task = await client.get_task("task-id", branch_name="main")
for prompt in task.prompts:
print(f"Prompt: {prompt.name}")
```
### get\_branch\_head
Get the head commit for a branch. Useful for determining deployment versions.
```python theme={null}
async def get_branch_head(
task_id: str,
branch_name: str = "main"
) -> BranchHeadResponse
```
**Returns:** `BranchHeadResponse` with:
* `effective_commit_id`: The commit ID to use
* `has_uncommitted_changes`: Whether there are pending changes
* `last_committed_at`: Timestamp of last commit
***
## Creating Sessions
### create\_prompt\_session
Create a new prompt session by fetching a prompt and combining it with session data.
```python theme={null}
async def create_prompt_session(
prompt_id: str,
branch_name: str | None = None,
commit_id: str | None = None,
session_data: RenderableModel | None = None,
) -> PromptSession
```
**Parameters:**
| Name | Type | Description |
| -------------- | ------------------------- | ---------------------------------------- |
| `prompt_id` | `str` | The prompt's anchor ID |
| `branch_name` | `str \| None` | Branch name |
| `commit_id` | `str \| None` | Commit ID |
| `session_data` | `RenderableModel \| None` | Pydantic model for variable substitution |
**Returns:** `PromptSession`
```python theme={null}
from models.my_task_models import QueryInput
session = await client.create_prompt_session(
prompt_id="prompt-id",
branch_name="main",
session_data=QueryInput(
query="What is the weather?",
user_name="Alice"
)
)
```
### prompt\_session\_from\_session\_data
Create a session using the prompt ID embedded in the session data's metadata.
```python theme={null}
async def prompt_session_from_session_data(
session_data: RenderableModel,
branch_name: str | None = None,
commit_id: str | None = None,
) -> PromptSession
```
This method reads the `prompt_id` from `session_data.moxn_schema_metadata`, so you don't need to specify it separately.
```python theme={null}
# Session data knows which prompt it belongs to
session = await client.prompt_session_from_session_data(
session_data=QueryInput(query="Hello"),
branch_name="main"
)
```
***
## Telemetry & Spans
### span
Create a span for tracing and observability.
```python theme={null}
@asynccontextmanager
async def span(
prompt_session: PromptSession,
name: str | None = None,
metadata: dict[str, Any] | None = None,
*,
parent_context: SpanContext | None = None,
trace_context: TraceContext | None = None,
) -> AsyncGenerator[Span, None]
```
**Parameters:**
| Name | Type | Description |
| ---------------- | ---------------------- | ------------------------------------------------------- |
| `prompt_session` | `PromptSession` | The session for this span |
| `name` | `str \| None` | Span name (defaults to prompt name) |
| `metadata` | `dict \| None` | Searchable attributes (customer\_id, request\_id, etc.) |
| `parent_context` | `SpanContext \| None` | Explicit parent for async patterns |
| `trace_context` | `TraceContext \| None` | For distributed tracing |
**Returns:** `AsyncGenerator[Span, None]`
```python theme={null}
async with client.span(
session,
name="process_query",
metadata={
"customer_id": "cust_123",
"status": "processing"
}
) as span:
response = anthropic.messages.create(**session.to_anthropic_invocation())
await client.log_telemetry_event_from_response(session, response, Provider.ANTHROPIC)
```
### span\_from\_carrier
Create a span from a trace carrier for distributed tracing.
```python theme={null}
@asynccontextmanager
async def span_from_carrier(
carrier: MoxnTraceCarrier,
name: str | None = None,
metadata: dict[str, Any] | None = None,
) -> AsyncGenerator[Span, None]
```
Use this in queue workers or callback handlers to continue a trace.
```python theme={null}
# In a queue worker
carrier = MoxnTraceCarrier.model_validate(message["carrier"])
async with client.span_from_carrier(carrier, name="process_item") as span:
await process(message["data"])
```
### extract\_context
Extract the current span context for propagation across services.
```python theme={null}
def extract_context() -> MoxnTraceCarrier | None
```
**Returns:** `MoxnTraceCarrier` if there's an active span, `None` otherwise.
```python theme={null}
async with client.span(session) as span:
carrier = client.extract_context()
if carrier:
await queue.put({
"carrier": carrier.model_dump(mode="json"),
"data": payload
})
```
### log\_telemetry\_event
Log an LLM event to the current span.
```python theme={null}
async def log_telemetry_event(event: LLMEvent) -> None
```
```python theme={null}
event = session.create_llm_event_from_parsed_response(parsed_response)
await client.log_telemetry_event(event)
```
### log\_telemetry\_event\_from\_response
Convenience method to parse a provider response and log it in one call.
```python theme={null}
async def log_telemetry_event_from_response(
prompt_session: PromptSession,
response: AnthropicMessage | OpenAIChatCompletion | GoogleGenerateContentResponse,
provider: Provider,
) -> None
```
**Parameters:**
| Name | Type | Description |
| ---------------- | ----------------- | ---------------------------------- |
| `prompt_session` | `PromptSession` | The session used for the request |
| `response` | Provider response | Raw response from the LLM provider |
| `provider` | `Provider` | The provider enum value |
```python theme={null}
response = anthropic.messages.create(**session.to_anthropic_invocation())
await client.log_telemetry_event_from_response(session, response, Provider.ANTHROPIC)
```
### flush
Await in-flight telemetry logs. Call this at process exit or Lambda return.
```python theme={null}
async def flush(timeout: float | None = None) -> None
```
```python theme={null}
# At shutdown
await client.flush(timeout=5.0)
```
***
## Code Generation
### generate\_task\_models
Generate Pydantic models from all schemas in a task.
```python theme={null}
async def generate_task_models(
task_id: str | UUID,
branch_name: str | None = None,
commit_id: str | None = None,
output_dir: Path | str | None = None,
) -> DatamodelCodegenResponse
```
**Parameters:**
| Name | Type | Description |
| ------------- | --------------------- | --------------------------------- |
| `task_id` | `str \| UUID` | The task ID |
| `branch_name` | `str \| None` | Branch name (defaults to "main") |
| `commit_id` | `str \| None` | Commit ID |
| `output_dir` | `Path \| str \| None` | Directory to write generated code |
**Returns:** `DatamodelCodegenResponse` with generated code
```python theme={null}
# Generate and save models
response = await client.generate_task_models(
task_id="task-id",
branch_name="main",
output_dir="./models"
)
print(f"Generated: {response.filename}")
```
***
## Task & Prompt Creation
### create\_task
Create a new task programmatically.
```python theme={null}
async def create_task(
name: str,
description: str | None = None,
branch_name: str = "main",
) -> Task
```
```python theme={null}
task = await client.create_task(
name="Customer Support Bot",
description="Handles customer inquiries"
)
```
### create\_prompt
Create a new prompt with messages.
```python theme={null}
async def create_prompt(
task_id: str | UUID,
name: str,
messages: list[MessageData | dict],
description: str | None = None,
branch_name: str = "main",
) -> PromptTemplate
```
```python theme={null}
from moxn.types.requests import MessageData
from moxn.types.content import MessageRole
from moxn.types.blocks.text import TextContentModel
messages = [
MessageData(
name="System",
role=MessageRole.SYSTEM,
blocks=[[TextContentModel(text="You are helpful.")]]
)
]
prompt = await client.create_prompt(
task_id=task.id,
name="Q&A Prompt",
messages=messages
)
```
***
## Lifecycle Methods
### close
Close the underlying HTTP client.
```python theme={null}
async def close() -> None
```
### verify\_access
Verify API key and return identity information.
```python theme={null}
async def verify_access() -> dict
```
```python theme={null}
info = await client.verify_access()
print(f"Tenant: {info['tenant_id']}")
```
# PromptSession
Source: https://moxn.mintlify.app/reference/session
API reference for the PromptSession class
The `PromptSession` is the runtime representation of a prompt template combined with session data. It handles variable substitution, provider conversion, and response parsing.
## Creation
Sessions are created from prompt templates using the client:
```python theme={null}
session = await client.create_prompt_session(
prompt_id="prompt-id",
branch_name="main",
session_data=QueryInput(query="Hello")
)
```
Or directly from a template:
```python theme={null}
session = PromptSession.from_prompt_template(
prompt=prompt_template,
session_data=QueryInput(query="Hello")
)
```
## Properties
| Property | Type | Description |
| -------------- | ------------------------- | ------------------------------------------ |
| `id` | `UUID` | Unique session identifier |
| `prompt` | `PromptTemplate` | The underlying prompt template |
| `prompt_id` | `UUID` | Shortcut to `prompt.id` |
| `messages` | `list[Message]` | Current message list |
| `session_data` | `RenderableModel \| None` | The session data for variable substitution |
***
## Provider Conversion
These methods convert the session to provider-specific formats for direct SDK usage.
### to\_anthropic\_invocation
Convert to complete Anthropic API payload with model parameters.
```python theme={null}
def to_anthropic_invocation(
context: MessageContext | dict | None = None,
*,
model: str | None = None,
max_tokens: int | None = None,
temperature: float | None = None,
top_p: float | None = None,
thinking: dict[str, Any] | None = None,
) -> AnthropicInvocationParam
```
**Parameters:**
| Name | Type | Description |
| ------------- | -------------------------------- | ------------------------------------------------- |
| `context` | `MessageContext \| dict \| None` | Additional context (merged with session\_data) |
| `model` | `str \| None` | Model override (e.g., "claude-sonnet-4-20250514") |
| `max_tokens` | `int \| None` | Max tokens override |
| `temperature` | `float \| None` | Temperature override |
| `top_p` | `float \| None` | Top-p override |
| `thinking` | `dict \| None` | Extended thinking configuration |
**Returns:** Complete payload ready for `**` unpacking.
```python theme={null}
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# With overrides
response = anthropic.messages.create(
**session.to_anthropic_invocation(
model="claude-sonnet-4-20250514",
max_tokens=1024
)
)
```
**Automatic inclusions:**
* Tools (if configured in prompt)
* Structured output format (if response schema configured)
* Model parameters from `completion_config`
For structured outputs, add the beta header manually:
```python theme={null}
extra_headers={"anthropic-beta": "structured-outputs-2025-11-13"}
```
### to\_openai\_chat\_invocation
Convert to complete OpenAI Chat API payload.
```python theme={null}
def to_openai_chat_invocation(
context: MessageContext | dict | None = None,
*,
model: str | None = None,
max_tokens: int | None = None,
temperature: float | None = None,
top_p: float | None = None,
thinking: dict[str, Any] | None = None,
) -> OpenAIChatInvocationParam
```
```python theme={null}
response = openai.chat.completions.create(
**session.to_openai_chat_invocation()
)
```
### to\_openai\_responses\_invocation
Convert to OpenAI Responses API payload.
```python theme={null}
def to_openai_responses_invocation(
context: MessageContext | dict | None = None,
*,
model: str | None = None,
max_tokens: int | None = None,
temperature: float | None = None,
top_p: float | None = None,
thinking: dict[str, Any] | None = None,
) -> OpenAIResponsesInvocationParam
```
```python theme={null}
response = openai.responses.create(
**session.to_openai_responses_invocation()
)
```
### to\_google\_gemini\_invocation
Convert to Google Gemini API payload.
```python theme={null}
def to_google_gemini_invocation(
context: MessageContext | dict | None = None,
*,
model: str | None = None,
max_tokens: int | None = None,
temperature: float | None = None,
top_p: float | None = None,
thinking: dict[str, Any] | None = None,
) -> GoogleInvocationParam
```
```python theme={null}
response = client.models.generate_content(
**session.to_google_gemini_invocation()
)
```
### to\_google\_vertex\_invocation
Convert to Google Vertex AI payload.
```python theme={null}
def to_google_vertex_invocation(
context: MessageContext | dict | None = None,
*,
model: str | None = None,
max_tokens: int | None = None,
temperature: float | None = None,
top_p: float | None = None,
thinking: dict[str, Any] | None = None,
) -> GoogleInvocationParam
```
### to\_invocation
Generic method that auto-selects provider based on `completion_config.provider`.
```python theme={null}
def to_invocation(
context: MessageContext | dict | None = None,
*,
provider: Provider | None = None,
model: str | None = None,
max_tokens: int | None = None,
temperature: float | None = None,
top_p: float | None = None,
thinking: dict[str, Any] | None = None,
) -> ProviderInvocationPayload
```
***
## Message-Only Conversion
These methods return just the messages without model parameters.
### to\_anthropic\_messages
```python theme={null}
def to_anthropic_messages(
context: MessageContext | dict | None = None
) -> AnthropicMessagesParam
```
Returns `{"system": ..., "messages": [...]}`.
### to\_openai\_chat\_messages
```python theme={null}
def to_openai_chat_messages(
context: MessageContext | dict | None = None
) -> OpenAIChatMessagesParam
```
Returns `{"messages": [...]}`.
### to\_openai\_responses\_messages
```python theme={null}
def to_openai_responses_messages(
context: MessageContext | dict | None = None
) -> OpenAIResponsesMessagesParam
```
Returns `{"input": [...], "instructions": ...}`.
### to\_google\_gemini\_messages
```python theme={null}
def to_google_gemini_messages(
context: MessageContext | dict | None = None
) -> GoogleMessagesParam
```
Returns `{"system_instruction": ..., "content": [...]}`.
### to\_messages
Generic method for any provider. Uses stored `completion_config.provider` by default:
```python theme={null}
def to_messages(
provider: Provider | None = None,
context: MessageContext | dict | None = None,
) -> ProviderMessagesParam
```
```python theme={null}
# Uses stored provider from completion_config
messages = session.to_messages()
# Or override explicitly
messages = session.to_messages(provider=Provider.ANTHROPIC)
```
### to\_payload
Generic method for any provider. Uses stored `completion_config.provider` by default:
```python theme={null}
def to_payload(
provider: Provider | None = None,
context: MessageContext | dict | None = None,
) -> ProviderPayload
```
```python theme={null}
# Uses stored provider from completion_config
payload = session.to_payload()
# Or override explicitly
payload = session.to_payload(provider=Provider.ANTHROPIC)
```
***
## Response Handling
### parse\_response
Parse a provider response into normalized format. Uses stored `completion_config.provider` by default:
```python theme={null}
def parse_response(
response: Any,
provider: Provider | None = None,
) -> ParsedResponse
```
```python theme={null}
response = anthropic.messages.create(**session.to_anthropic_invocation())
# Uses stored provider from completion_config
parsed = session.parse_response(response)
# Or override explicitly
parsed = session.parse_response(response, provider=Provider.ANTHROPIC)
# Access normalized content
for candidate in parsed.candidates:
for block in candidate.content_blocks:
if isinstance(block, TextContent):
print(block.text)
```
### create\_llm\_event\_from\_response
Create an LLM event for telemetry from a raw provider response. Uses stored `completion_config.provider` by default:
```python theme={null}
def create_llm_event_from_response(
response: AnthropicMessage | OpenAIChatCompletion | GoogleGenerateContentResponse,
provider: Provider | None = None,
) -> LLMEvent
```
```python theme={null}
response = anthropic.messages.create(**session.to_anthropic_invocation())
# Uses stored provider from completion_config
event = session.create_llm_event_from_response(response)
await client.log_telemetry_event(event)
# Or override explicitly
event = session.create_llm_event_from_response(response, provider=Provider.ANTHROPIC)
```
### create\_llm\_event\_from\_parsed\_response
Create an LLM event from an already-parsed response with additional metadata.
```python theme={null}
def create_llm_event_from_parsed_response(
parsed_response: ParsedResponse,
request_config: RequestConfig | None = None,
schema_definition: SchemaDefinition | None = None,
attributes: dict[str, Any] | None = None,
validation_errors: list[str] | None = None,
) -> LLMEvent
```
**Parameters:**
| Name | Type | Description |
| ------------------- | -------------------------- | --------------------------------------------- |
| `parsed_response` | `ParsedResponse` | The parsed response |
| `request_config` | `RequestConfig \| None` | Request configuration used |
| `schema_definition` | `SchemaDefinition \| None` | Schema or tool definitions |
| `attributes` | `dict \| None` | Custom attributes |
| `validation_errors` | `list[str] \| None` | Validation errors if schema validation failed |
```python theme={null}
parsed = session.parse_response(response)
# Validate structured output
try:
result = MySchema.model_validate_json(parsed.candidates[0].content[0].text)
validation_errors = None
except ValidationError as e:
validation_errors = [str(err) for err in e.errors()]
result = None
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
validation_errors=validation_errors
)
await client.log_telemetry_event(event)
```
***
## Runtime Message Management
Add messages to the session at runtime for multi-turn conversations.
### append\_user\_text
Append a user message with text content.
```python theme={null}
def append_user_text(
text: str,
name: str = "",
description: str = "",
) -> None
```
```python theme={null}
session.append_user_text("What about the weather tomorrow?")
```
### append\_assistant\_text
Append an assistant message with text content.
```python theme={null}
def append_assistant_text(
text: str,
name: str = "",
description: str = "",
) -> None
```
```python theme={null}
session.append_assistant_text("The weather will be sunny.")
```
### append\_assistant\_response
Append an assistant response from a parsed LLM response.
```python theme={null}
def append_assistant_response(
parsed_response: ParsedResponse,
candidate_idx: int = 0,
name: str = "",
description: str = "",
) -> None
```
This is the preferred way to add LLM responses for multi-turn conversations:
```python theme={null}
# First turn
response = anthropic.messages.create(**session.to_anthropic_invocation())
parsed = session.parse_response(response)
session.append_assistant_response(parsed)
# Second turn
session.append_user_text("Tell me more")
response = anthropic.messages.create(**session.to_anthropic_invocation())
```
***
## Context Handling
The session automatically handles context from `session_data`:
1. If `session_data` is provided, `render()` is called to get variables
2. These variables are used for substitution in message templates
3. Additional `context` parameters are merged (taking precedence)
```python theme={null}
# session_data provides base variables
session = await client.create_prompt_session(
prompt_id="prompt-id",
session_data=QueryInput(query="Hello", user="Alice")
)
# Additional context overrides session_data
payload = session.to_anthropic_invocation(
context={"user": "Bob"} # Overrides session_data.user
)
```
# Types Reference
Source: https://moxn.mintlify.app/reference/types
Key types used throughout the Moxn SDK
This page documents the key types you'll encounter when using the Moxn SDK.
## Provider
Enum representing supported LLM providers.
```python theme={null}
from moxn.types.content import Provider
class Provider(Enum):
ANTHROPIC = "anthropic"
OPENAI_CHAT = "openai_chat"
OPENAI_RESPONSES = "openai_responses"
GOOGLE_GEMINI = "google_gemini"
GOOGLE_VERTEX = "google_vertex"
```
**Usage:**
```python theme={null}
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
***
## RenderableModel
Protocol for session data models generated by codegen.
```python theme={null}
from moxn.types.base import RenderableModel
@runtime_checkable
class RenderableModel(Protocol):
moxn_schema_metadata: ClassVar[MoxnSchemaMetadata]
def model_dump(self, ...) -> Any: ...
def render(self, **kwargs: Any) -> Any: ...
```
Generated models implement this protocol:
```python theme={null}
class QueryInput(BaseModel):
"""Generated by Moxn codegen."""
moxn_schema_metadata: ClassVar[MoxnSchemaMetadata] = MoxnSchemaMetadata(
schema_id=UUID("..."),
prompt_id=UUID("..."),
task_id=UUID("...")
)
query: str
context: str | None = None
def render(self, **kwargs) -> dict[str, str]:
return {
"query": self.query,
"context": self.context or ""
}
```
The `render()` method transforms typed data into a flat `dict[str, str]` for variable substitution.
***
## PromptTemplate
A prompt template fetched from the Moxn API.
```python theme={null}
class PromptTemplate:
id: UUID # Stable anchor ID
name: str # Prompt name
description: str | None # Optional description
task_id: UUID # Parent task ID
messages: list[Message] # Ordered messages
input_schema: Schema | None # Auto-generated input schema
completion_config: CompletionConfig | None # Model settings
tools: list[SdkTool] | None # Tools and structured outputs
branch_id: UUID | None # Branch (if branch access)
commit_id: UUID | None # Commit (if commit access)
```
**Key Properties:**
| Property | Type | Description |
| -------------------------- | -------------------------- | ------------------------------------- |
| `id` | `UUID` | Stable anchor ID that never changes |
| `name` | `str` | Human-readable prompt name |
| `messages` | `list[Message]` | The message sequence |
| `input_schema` | `Schema \| None` | Auto-generated from variables |
| `completion_config` | `CompletionConfig \| None` | Model and parameter settings |
| `function_tools` | `list[SdkTool]` | Tools configured for function calling |
| `structured_output_schema` | `SdkTool \| None` | Schema for structured output |
***
## Task
A task containing prompts and schemas.
```python theme={null}
class Task:
id: UUID # Stable anchor ID
name: str # Task name
description: str | None # Optional description
prompts: list[PromptTemplate] # All prompts in task
definitions: dict[str, Any] # Schema definitions
branches: list[Branch] # All branches
last_commit: Commit | None # Latest commit info
branch_id: UUID | None # Current branch
commit_id: UUID | None # Current commit
```
***
## Message
A message within a prompt.
```python theme={null}
class Message:
id: UUID # Stable anchor ID
name: str # Message name
description: str | None # Optional description
role: MessageRole # system, user, assistant
author: Author # HUMAN or MACHINE
blocks: list[list[ContentBlock]] # 2D array of content blocks
task_id: UUID # Parent task ID
```
**Roles:**
```python theme={null}
class MessageRole(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
```
***
## ParsedResponse
Normalized LLM response from any provider.
```python theme={null}
class ParsedResponse:
provider: Provider # Which provider
candidates: list[ParsedResponseCandidate] # Response candidates
stop_reason: StopReason # Why generation stopped
usage: TokenUsage # Token counts
model: str | None # Model used
raw_response: dict # Original response
```
### ParsedResponseCandidate
A single response candidate.
```python theme={null}
class ParsedResponseCandidate:
content_blocks: list[TextContent | ToolCall | ThinkingContent | ReasoningContent]
metadata: ResponseMetadata
```
Content blocks preserve order, which is important for interleaved thinking/text/tool sequences.
### Content Block Types
```python theme={null}
class TextContent(BaseModel):
text: str
class ToolCall(BaseModel):
id: str
name: str
arguments: dict[str, Any] | str | None
class ThinkingContent(BaseModel):
thinking: str # For Claude extended thinking
class ReasoningContent(BaseModel):
summary: str # For OpenAI o1/o3 reasoning
```
### StopReason
```python theme={null}
class StopReason(str, Enum):
END_TURN = "end_turn"
MAX_TOKENS = "max_tokens"
TOOL_CALL = "tool_call"
CONTENT_FILTER = "content_filter"
ERROR = "error"
OTHER = "other"
```
### TokenUsage
```python theme={null}
class TokenUsage(BaseModel):
input_tokens: int | None = None
completion_tokens: int | None = None
thinking_tokens: int | None = None # For extended thinking models
```
***
## LLMEvent
Event logged to telemetry for LLM interactions.
```python theme={null}
class LLMEvent:
promptId: UUID
promptName: str
taskId: UUID
branchId: UUID | None
commitId: UUID | None
messages: list[Message] # Messages sent
provider: Provider
rawResponse: dict # Original response
parsedResponse: ParsedResponse # Normalized response
sessionData: RenderableModel | None # Input data (typed)
renderedInput: dict[str, str] | None # Rendered variables (flat)
attributes: dict[str, Any] | None # Custom attributes
isUncommitted: bool # Whether from uncommitted state
responseType: ResponseType # Classification
validationErrors: list[str] | None # Schema validation errors
```
### ResponseType
Classification of the response for UI rendering:
```python theme={null}
class ResponseType(str, Enum):
TEXT = "text"
TOOL_CALLS = "tool_calls"
TEXT_WITH_TOOLS = "text_with_tools"
STRUCTURED = "structured"
STRUCTURED_WITH_TOOLS = "structured_with_tools"
THINKING = "thinking"
TEXT_WITH_THINKING = "text_with_thinking"
THINKING_WITH_TOOLS = "thinking_with_tools"
```
***
## Span Types
### Span
An active span for tracing.
```python theme={null}
class Span:
context: SpanContext # Context for propagation
def set_attribute(self, key: str, value: Any) -> None:
"""Add searchable metadata to the span."""
```
### SpanContext
Context that can be passed to child spans.
```python theme={null}
class SpanContext:
trace_id: str
span_id: str
```
### MoxnTraceCarrier
Carrier for propagating trace context across services.
```python theme={null}
class MoxnTraceCarrier(BaseModel):
traceparent: str # W3C trace context
tracestate: str | None
moxn_metadata: dict # Moxn-specific context
```
**Usage for distributed tracing:**
```python theme={null}
# Extract from current span
carrier = client.extract_context()
# Send to another service
await queue.put({"carrier": carrier.model_dump(mode="json"), "data": ...})
# In the receiving service
carrier = MoxnTraceCarrier.model_validate(message["carrier"])
async with client.span_from_carrier(carrier) as span:
await process(message["data"])
```
***
## Version Types
### VersionRef
Reference to a specific version (branch or commit).
```python theme={null}
class VersionRef(BaseModel):
branch_name: str | None = None
commit_id: str | None = None
```
Exactly one of `branch_name` or `commit_id` must be provided.
### BranchHeadResponse
Response from `get_branch_head()`.
```python theme={null}
class BranchHeadResponse(BaseModel):
branch_id: str
branch_name: str
task_id: UUID
head_commit_id: str | None
effective_commit_id: str # The commit to use
has_uncommitted_changes: bool
last_committed_at: datetime | None
is_default: bool
```
### Branch
A branch reference.
```python theme={null}
class Branch(BaseModel):
id: UUID
name: str
head_commit_id: UUID | None
```
### Commit
A commit snapshot.
```python theme={null}
class Commit(BaseModel):
id: str # Commit SHA
message: str
created_at: datetime | None
```
***
## Schema Types
### MoxnSchemaMetadata
Metadata embedded in JSON schemas.
```python theme={null}
class MoxnSchemaMetadata(BaseModel):
schema_id: UUID
schema_version_id: UUID | None
prompt_id: UUID | None # None for task-level schemas
prompt_version_id: UUID | None
task_id: UUID
branch_id: UUID | None
commit_id: str | None
```
This metadata is included in the `x-moxn-metadata` field of exported JSON schemas and in generated Pydantic models.
# Logging Events
Source: https://moxn.mintlify.app/telemetry/logging
Log LLM interactions with full context
Every LLM call should be logged for debugging, analysis, and compliance. This guide covers how to log telemetry events with Moxn.
## Basic Logging
The simplest way to log an LLM call:
```python theme={null}
async with client.span(session) as span:
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# Log with provider type
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
This logs:
* The complete prompt (all messages)
* Your session data and rendered input
* The raw LLM response
* Token counts and model info
* Span context for tracing
## What Gets Logged
Each telemetry event captures:
### Session Data
Your original typed input:
```python theme={null}
{
"query": "How do I reset my password?",
"customer_name": "Alice",
"documents": [...] # Original Pydantic objects
}
```
### Rendered Input
The flattened strings injected into the prompt:
```python theme={null}
{
"query": "How do I reset my password?",
"customer_name": "Alice",
"documents": "[{\"title\": \"FAQ\", ...}]" # Serialized
}
```
### Messages
The complete prompt sent to the LLM:
```python theme={null}
[
{"role": "system", "content": "You are a helpful assistant..."},
{"role": "user", "content": "Customer Alice asks: How do I..."}
]
```
### Response
The raw LLM response plus parsed content:
```python theme={null}
{
"raw_response": {...}, # Original provider response
"parsed_response": {
"candidates": [...],
"input_tokens": 150,
"output_tokens": 200,
"model": "claude-sonnet-4-20250514",
"stop_reason": "end_turn"
}
}
```
### Metadata
Context about the call:
```python theme={null}
{
"prompt_id": "uuid",
"prompt_name": "Product Help",
"task_id": "uuid",
"branch_id": "uuid", # If branch access
"commit_id": "abc123", # If commit access
"provider": "anthropic",
"is_uncommitted": false
}
```
## Logging Methods
### log\_telemetry\_event\_from\_response()
The recommended method—handles parsing automatically:
```python theme={null}
# Anthropic
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
# OpenAI
await client.log_telemetry_event_from_response(
session, response, Provider.OPENAI_CHAT
)
# Google
await client.log_telemetry_event_from_response(
session, response, Provider.GOOGLE_GEMINI
)
```
### log\_telemetry\_event()
For manual event creation with more control:
```python theme={null}
# Create event manually
event = session.create_llm_event_from_response(
response=response,
provider=Provider.ANTHROPIC
)
# Or from parsed response with extra data
parsed = session.parse_response(response)
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
request_config=config, # Optional
schema_definition=schema, # Optional
attributes={"custom": "data"}, # Optional
validation_errors=["field missing"] # Optional
)
# Log it
await client.log_telemetry_event(event)
```
## LLMEvent Structure
The `LLMEvent` contains all logged data:
```python theme={null}
event = session.create_llm_event_from_response(response, Provider.ANTHROPIC)
# Prompt context
event.promptId # UUID
event.promptName # str
event.taskId # UUID
event.branchId # UUID | None
event.commitId # UUID | None
event.isUncommitted # bool
# Input
event.messages # list[Message] - the prompt
event.sessionData # RenderableModel | None
event.renderedInput # dict[str, str] | None
# Output
event.provider # Provider enum
event.rawResponse # dict
event.parsedResponse # ParsedResponse
# Enhanced telemetry
event.responseType # str - "text", "tool_call", etc.
event.requestConfig # RequestConfig | None
event.schemaDefinition # SchemaDefinition | None
event.toolCallsCount # int
event.validationErrors # list[str] | None
# Custom data
event.attributes # dict | None
```
## Custom Attributes
Add custom data to events by setting metadata when creating the span:
```python theme={null}
# Set metadata at span creation (preferred)
async with client.span(
session,
metadata={
"customer_id": "123",
"query_type": "password_reset",
"customer_tier": "premium",
"ab_test_variant": "B"
}
) as span:
response = await call_llm()
await client.log_telemetry_event_from_response(...)
```
This metadata is captured with all telemetry events logged within the span.
For event-specific attributes, use `create_llm_event_from_parsed_response`:
```python theme={null}
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
attributes={
"validation_passed": True,
"response_category": "helpful"
}
)
```
## Validation Logging
Log validation results for structured outputs:
```python theme={null}
from pydantic import ValidationError
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
parsed = session.parse_response(response)
# Attempt to validate structured output
validation_errors = []
try:
result = YourOutputModel.model_validate_json(
parsed.candidates[0].content[0].text
)
except ValidationError as e:
validation_errors = [str(err) for err in e.errors()]
# Log with validation status
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
validation_errors=validation_errors if validation_errors else None
)
await client.log_telemetry_event(event)
```
## Event Types
Events are categorized by type:
| Type | Description |
| ------------------- | ---------------------------- |
| `text` | Standard text response |
| `tool_call` | Response includes tool calls |
| `structured_output` | JSON structured output |
| `thinking` | Extended thinking response |
| `mixed` | Multiple content types |
The SDK detects the type automatically:
```python theme={null}
event = session.create_llm_event_from_response(response, Provider.ANTHROPIC)
print(event.responseType) # "text", "tool_call", etc.
```
## Logging Without Spans
You can log events without a span context, but it's not recommended:
```python theme={null}
# Works but loses trace hierarchy
response = anthropic.messages.create(...)
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
The event will still be logged, but won't be part of a trace hierarchy.
## Batch Processing
For batch operations, log each item individually:
```python theme={null}
async with client.span(session, name="batch_process") as root_span:
for i, item in enumerate(items):
async with client.span(
session,
name=f"process_item_{i}",
metadata={"item_id": item.id}
) as span:
response = await process_item(item)
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
## Error Logging
Spans automatically capture errors when exceptions are raised:
```python theme={null}
async with client.span(
session,
metadata={"operation": "llm_call"}
) as span:
try:
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
except anthropic.RateLimitError as e:
# The span automatically captures error.type and error.message
# when the exception propagates
raise
```
## Token Tracking
Token counts are extracted automatically:
```python theme={null}
event = session.create_llm_event_from_response(response, Provider.ANTHROPIC)
print(event.parsedResponse.input_tokens) # Prompt tokens
print(event.parsedResponse.output_tokens) # Completion tokens
```
These appear in the web app for cost analysis.
## Complete Example
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
from pydantic import ValidationError
async def analyze_with_logging(query: str):
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="analysis-prompt",
session_data=AnalysisInput(query=query)
)
# Set all known metadata at span creation
async with client.span(
session,
name="structured_analysis",
metadata={
"query_length": len(query),
"query_category": classify_query(query)
}
) as span:
# Make the LLM call
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# Parse response (uses stored provider from completion_config)
parsed = session.parse_response(response)
# Validate structured output
validation_errors = []
try:
result = AnalysisOutput.model_validate_json(
parsed.candidates[0].content[0].text
)
except ValidationError as e:
validation_errors = [str(err) for err in e.errors()]
result = None
# Log the complete event with validation status
event = session.create_llm_event_from_parsed_response(
parsed_response=parsed,
validation_errors=validation_errors or None,
attributes={
"output_valid": result is not None,
"result_confidence": result.confidence if result else None
}
)
await client.log_telemetry_event(event)
return result
```
## Next Steps
View logs in the web app
Create trace hierarchies
Event creation methods
See logging in context
# Reviewing Telemetry
Source: https://moxn.mintlify.app/telemetry/reviewing
View and analyze traces in the Moxn web app
After logging telemetry, you can review it in the Moxn web app. This guide shows you how to find and analyze your traces.
## Accessing Traces
Navigate to your task in the Moxn web app and open the **Traces** tab.
You'll see a list of all traces for that task, showing:
* Timestamp
* Prompt name
* Duration
* Token count
* Status
## Traces Tab
![Traces list view]()
The Traces tab shows all executions for your task with filtering controls at the top.
## Trace List View
The trace list provides a high-level overview:
| Column | Description |
| ------------ | --------------------------------------- |
| **Time** | When the trace started |
| **Prompt** | Which prompt was used |
| **Duration** | Total time for the trace |
| **Tokens** | Input + output tokens |
| **Cost** | Estimated cost (based on model pricing) |
| **Status** | Success, error, or warning |
### Filtering
Filter traces by:
* **Date range**: Today, last 7 days, custom range
* **Prompt**: Filter to specific prompts
* **Status**: Success, error
* **Branch/Commit**: See traces from specific versions
### Sorting
Sort by:
* Newest/oldest
* Duration (slowest first)
* Token count (highest first)
* Cost (highest first)
## Trace Detail View
Click a trace to see the full details:
![Trace detail with span hierarchy]()
### Trace Header
The header shows:
* Trace name
* Summary stats (completions, tool calls, token counts)
* Show details button
### Span Hierarchy
See the tree structure of spans:
* **customer\_support\_request** (root) - 2.5s
* classify\_query - 0.8s
* search\_documents - 0.3s
* generate\_response - 1.4s
* LLM Call
Click any span to see its details.
### Span Details
For each span, you can see:
**Timing**
* Start time
* Duration
* Percentage of total trace time
**Attributes**
* Custom metadata you added
* System attributes (prompt\_id, task\_id, etc.)
**Events**
* LLM calls made within the span
* Token counts per event
* Response types
## Span Detail Modal
Click any span to view detailed information:
![Span detail modal]()
### Modal Tabs
| Tab | Content |
| ----------------------- | -------------------------------------------- |
| **Conversation Flow** | Visual message sequence with role indicators |
| **Variables** | Input variable values used in this call |
| **Metrics** | Latency, token usage, estimated costs |
| **Raw Message Content** | Full message content |
| **Raw Data** | Complete span data as JSON |
### Navigation
* **Previous/Next**: Navigate between spans
* **Keyboard shortcuts**: Arrow keys, J/K, Esc to close
* **Create Observation**: Save span for test cases
## LLM Event Details
Click an LLM event to see the complete interaction:
### Input Tab
**Session Data**: Your original typed input
```json theme={null}
{
"query": "How do I reset my password?",
"customer_name": "Alice",
"documents": [
{"title": "FAQ", "content": "..."}
]
}
```
**Rendered Input**: The flattened values used for substitution
```json theme={null}
{
"query": "How do I reset my password?",
"customer_name": "Alice",
"documents": "[{\"title\": \"FAQ\", ...}]"
}
```
**Messages**: The complete prompt sent to the LLM
Shows each message with:
* Role (system/user/assistant)
* Content (with variables substituted)
* Any images or files included
### Output Tab
**Response Content**: What the LLM returned
For text responses:
```
To reset your password, follow these steps:
1. Go to the login page...
```
For tool calls:
```json theme={null}
{
"tool": "search_knowledge_base",
"input": {"query": "password reset"}
}
```
For structured output:
```json theme={null}
{
"category": "account",
"subcategory": "password",
"confidence": 0.95
}
```
**Raw Response**: The complete provider response (expandable)
### Metrics Tab
* **Model**: Which model was used
* **Provider**: Anthropic, OpenAI, etc.
* **Input tokens**: Prompt token count
* **Output tokens**: Completion token count
* **Total tokens**: Combined count
* **Estimated cost**: Based on model pricing
* **Latency**: Time to first token, total time
* **Stop reason**: Why the model stopped
### Version Tab
* **Prompt ID**: UUID of the prompt
* **Prompt name**: Human-readable name
* **Branch**: If fetched by branch
* **Commit**: If fetched by commit
* **Uncommitted**: Whether working state was used
## Use Cases
### Debugging Issues
When something goes wrong:
1. Filter to the timeframe when the issue occurred
2. Find traces with errors
3. Click to see the full input/output
4. Check if:
* Input data was correct
* Prompt content was expected
* Response was malformed
### Performance Analysis
To identify slow traces:
1. Sort by duration (slowest first)
2. Look for patterns:
* Long input (too many tokens?)
* Slow model (use a faster one?)
* Sequential calls (could parallelize?)
3. Check span hierarchy for bottlenecks
### Cost Optimization
To reduce costs:
1. Sort by cost (highest first)
2. Identify expensive prompts
3. Look for:
* Unnecessarily long context
* Verbose system prompts
* Large documents that could be summarized
### A/B Testing
To compare versions:
1. Run both versions and log with different metadata
2. Filter by your A/B test attribute
3. Compare:
* Success rates
* Average latency
* Output quality (manual review)
## Exporting Data
Export trace data for external analysis:
* **CSV**: Basic metrics
* **JSON**: Full trace data
Use exports for:
* Building dashboards
* Long-term storage
* Compliance records
## Best Practices
Use custom attributes to make traces searchable:
```python theme={null}
metadata={"customer_id": "123", "feature": "password_reset"}
```
Use names that describe what the span does:
```python theme={null}
name="classify_customer_query" # Good
name="step_1" # Bad
```
Check traces periodically, not just when issues arise.
Use exports to build alerts for:
* Error rate spikes
* Latency increases
* Cost anomalies
## Next Steps
Create better trace hierarchies
Log more detailed events
See complex traces
Create prompts to trace
# Spans & Tracing
Source: https://moxn.mintlify.app/telemetry/spans
Observable traces for LLM workflows
Moxn provides W3C Trace Context compatible spans for full observability of your LLM workflows. This guide covers how to create spans and propagate trace context.
## Why Spans?
Spans give you:
* **Visibility**: See exactly what happened during an LLM call
* **Debugging**: Trace issues back to specific invocations
* **Analytics**: Measure latency, token usage, and costs
* **Correlation**: Link related LLM calls in complex workflows
## Basic Span Usage
Wrap LLM calls in spans to capture telemetry:
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="...",
session_data=your_input
)
# Create a span for this LLM call
async with client.span(session) as span:
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
# Log the event within the span
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
```
## Span Parameters
### Basic Parameters
```python theme={null}
async with client.span(
session,
name="analyze_query", # Custom span name (defaults to prompt name)
metadata={"key": "value"} # Searchable attributes
) as span:
...
```
### Full Signature
```python theme={null}
async with client.span(
prompt_session: PromptSession,
name: str | None = None,
metadata: dict[str, Any] | None = None,
*,
parent_context: SpanContext | None = None, # For async patterns
trace_context: TraceContext | None = None # For distributed tracing
) as span:
...
```
## Span Properties
Access span information within the context:
```python theme={null}
async with client.span(
session,
metadata={
"user_id": "user_123",
"query_type": "product_question"
}
) as span:
span.span_id # str - unique span identifier
span.context.trace_id # str - trace identifier
span.name # str - span name
span.parent_span_id # str | None - parent span (if nested)
span.start_time # datetime
span.attributes # dict - custom attributes (from metadata)
```
## Parent-Child Spans
Create hierarchical traces for complex workflows:
```python theme={null}
async with client.span(session, name="handle_request") as root_span:
# First child span
async with client.span(session, name="classify_query") as classify_span:
classification = await classify(query)
await client.log_telemetry_event_from_response(...)
# Second child span
async with client.span(session, name="generate_response") as response_span:
response = await generate(query, classification)
await client.log_telemetry_event_from_response(...)
```
This creates a trace hierarchy:
* **handle\_request** (root)
* classify\_query (child)
* generate\_response (child)
## Parallel Spans
For concurrent operations, pass parent context explicitly:
```python theme={null}
import asyncio
async with client.span(session, name="parallel_analysis") as root_span:
# Extract parent context for parallel tasks
root_context = root_span.context
async def analyze_sentiment():
async with client.span(
session,
name="sentiment_analysis",
parent_context=root_context # Explicit parent
) as span:
response = await call_llm_for_sentiment()
await client.log_telemetry_event_from_response(...)
return response
async def analyze_entities():
async with client.span(
session,
name="entity_extraction",
parent_context=root_context # Same parent
) as span:
response = await call_llm_for_entities()
await client.log_telemetry_event_from_response(...)
return response
# Run in parallel
sentiment, entities = await asyncio.gather(
analyze_sentiment(),
analyze_entities()
)
```
Result:
* **parallel\_analysis** (root)
* sentiment\_analysis (parallel child)
* entity\_extraction (parallel child)
## Distributed Tracing
Propagate traces across services using carriers:
### Extract Context
```python theme={null}
async with client.span(session, name="api_handler") as span:
# Extract context for propagation
carrier = client.extract_context()
if carrier:
# Send to another service via queue, HTTP, etc.
await queue.put({
"carrier": carrier.model_dump(mode="json"),
"payload": data
})
```
### Resume from Carrier
In another service or worker:
```python theme={null}
from moxn.types.telemetry import MoxnTraceCarrier
# Receive the carrier
message = await queue.get()
carrier = MoxnTraceCarrier.model_validate(message["carrier"])
# Create a span that continues the trace
async with client.span_from_carrier(
carrier,
name="process_async_task"
) as span:
# Process the work
result = await process(message["payload"])
await client.log_telemetry_event(...)
```
### Carrier Contents
```python theme={null}
carrier = client.extract_context()
carrier.trace_id # str - W3C trace ID
carrier.span_id # str - Parent span ID
carrier.prompt_id # UUID - Source prompt
carrier.prompt_name # str
carrier.task_id # UUID
carrier.commit_id # UUID | None
carrier.branch_id # UUID | None
```
## W3C Trace Context
Moxn spans are W3C Trace Context compatible:
```python theme={null}
# Extract W3C headers for HTTP propagation
async with client.span(session) as span:
headers = {
"traceparent": f"00-{span.context.trace_id}-{span.span_id}-01"
}
await http_client.post(url, headers=headers, json=data)
```
### Incoming HTTP Requests
Resume traces from incoming HTTP:
```python theme={null}
from moxn.types.telemetry import TraceContext
# Parse incoming traceparent header
traceparent = request.headers.get("traceparent")
if traceparent:
parts = traceparent.split("-")
trace_context = TraceContext(
trace_id=parts[1],
span_id=parts[2]
)
async with client.span(
session,
trace_context=trace_context # Continue existing trace
) as span:
...
```
## Adding Metadata
Add searchable metadata to spans by passing it in the `metadata` parameter at span creation:
```python theme={null}
async with client.span(
session,
metadata={
"customer_id": "cust_123",
"request_id": "req_456",
"environment": "production",
"query_type": "product_question",
"document_count": len(documents)
}
) as span:
response = await call_llm()
# Metadata is captured with all telemetry events in this span
```
This is the preferred pattern—define all known metadata upfront when creating the span context.
## Span Events
Within a span, you can log multiple events:
```python theme={null}
async with client.span(session, name="multi_step_workflow") as span:
# First LLM call
response1 = await call_llm_1()
await client.log_telemetry_event_from_response(
session, response1, Provider.ANTHROPIC
)
# Process result
processed = process(response1)
# Second LLM call
response2 = await call_llm_2(processed)
await client.log_telemetry_event_from_response(
session, response2, Provider.ANTHROPIC
)
```
Both events are associated with the same span.
## Error Handling
Spans capture errors automatically. Exceptions are recorded on the span when raised:
```python theme={null}
async with client.span(
session,
metadata={"operation": "llm_call"}
) as span:
try:
response = await call_llm()
await client.log_telemetry_event_from_response(...)
except Exception as e:
# The span automatically records exception details
# (error.type and error.message are captured)
raise
```
## Flushing Telemetry
Ensure all telemetry is sent before shutdown:
```python theme={null}
async with MoxnClient() as client:
# Your code...
# Explicit flush with timeout (optional)
await client.flush(timeout=5.0)
# Context manager automatically flushes on exit
```
For serverless or short-lived processes:
```python theme={null}
async def lambda_handler(event, context):
async with MoxnClient() as client:
session = await client.create_prompt_session(...)
async with client.span(session) as span:
response = await call_llm()
await client.log_telemetry_event_from_response(...)
# Ensure telemetry is sent before Lambda returns
await client.flush(timeout=3.0)
return {"statusCode": 200}
```
## Complete Example
```python theme={null}
from moxn import MoxnClient
from moxn.types.content import Provider
from anthropic import Anthropic
import asyncio
async def handle_customer_query(query: str, customer_id: str):
async with MoxnClient() as client:
session = await client.create_prompt_session(
prompt_id="product-help-prompt",
branch_name="main",
session_data=ProductHelpInput(query=query)
)
async with client.span(
session,
name="customer_support_request",
metadata={
"customer_id": customer_id,
"query_length": len(query)
}
) as root_span:
# Step 1: Classify the query
classification = await classify_query(query)
# Step 2: Search for relevant docs
docs = await search_knowledge_base(query)
# Step 3: Generate response with all context in metadata
async with client.span(
session,
name="generate",
metadata={
"classification": classification,
"doc_count": len(docs)
}
) as span:
anthropic = Anthropic()
response = anthropic.messages.create(
**session.to_anthropic_invocation()
)
await client.log_telemetry_event_from_response(
session, response, Provider.ANTHROPIC
)
return response.content[0].text
asyncio.run(handle_customer_query("How do I reset my password?", "cust_123"))
```
## Next Steps
Learn about event types and logging
View traces in the web app
See spans in a complex workflow
Full API documentation
# Part 2: Models and Context Preparation
Source: https://moxn.mintlify.app/videos/models
Video walkthrough of models and context preparation in Moxn
# Part 1: Web App Overview
Source: https://moxn.mintlify.app/videos/overview
Video walkthrough of the Moxn web application
# Part 3: Prompt Sessions, LLM Integration and Telemetry
Source: https://moxn.mintlify.app/videos/sessions
Video walkthrough of prompt sessions, LLM integration, and telemetry in Moxn
# Creating Tasks & Prompts
Source: https://moxn.mintlify.app/webapp/creating
Create and manage AI content in the Moxn web app
This guide walks through creating Tasks, Prompts, and Messages in the Moxn web app at [moxn.dev](https://moxn.dev).
## Dashboard Overview
When you log in, the dashboard shows your workspace:
![Moxn dashboard]()
The dashboard displays:
* **Tasks**: Your prompt collections
* **API Keys**: Manage keys for SDK access
* **Workspace info**: Team and usage details
## Creating a Task
A Task is your top-level container—think of it like a Git repository for a specific AI feature.
From the dashboard, click **Tasks** in the sidebar.
Click the **Create Task** button in the top right.
* **Name**: A descriptive name (e.g., "Customer Support Bot")
* **Description**: What this task is for
Click **Create** to create your task with a default "main" branch.
### Tasks List
![Tasks list]()
Each task card shows:
* Task name and description
* Last updated time
* Entity type badge
### Task Organization
Organize tasks by:
* **Feature**: One task per AI feature
* **Team**: One task per team's prompts
* **Project**: One task per project
* **Your Workspace**
* Customer Support Bot (Task)
* Search Assistant (Task)
* Document Analyzer (Task)
## Creating a Prompt
Prompts are templates for LLM invocations within a task.
Click on the task where you want to create the prompt.
Click **Create Prompt** or the **+** button.
* **Name**: The prompt's name (e.g., "Product Help")
* **Description**: What this prompt does
* **Folder**: Optional folder path (e.g., "support/tier1")
Click **Create** to add the prompt.
### Task Detail View
When you open a task, you see the **Prompts** tab:
![Task detail with prompts]()
The task detail page shows:
* **Prompts tab**: All prompts in this task
* **Schemas tab**: Input schemas and user-defined schemas
* **Traces tab**: Execution logs from SDK invocations
### Prompt Naming
Use folders to organize prompts:
* **Customer Support Bot**
* support/
* product-help
* billing-questions
* classification/
* query-classifier
* escalation/
* escalation-handler
The folder structure is stored in the prompt name itself.
## Adding Messages
Messages are the content blocks that make up your prompt.
Click on the prompt to open the editor.
Click **Add Message** or the **+** in the messages panel.
Select the message role:
* **System**: Instructions for the LLM
* **User**: User input template
* **Assistant**: Example responses (for few-shot)
Use the rich text editor to write your message content.
### Prompt Detail View
Opening a prompt shows its components:
![Prompt detail page]()
The prompt page displays:
* **Input Schema**: Auto-generated from message variables
* **Tools**: Attached schemas for function calling
* **Message Templates**: The ordered list of messages
### Message Types
| Role | Purpose | Example |
| ------------- | ------------------- | --------------------------------------------- |
| **System** | LLM instructions | "You are a helpful customer support agent..." |
| **User** | User input template | `Customer asks: {{query}}` |
| **Assistant** | Example output | "I'd be happy to help with that..." |
### Rich Content
The editor supports:
* **Text**: Plain text content
* **Variables**: Dynamic values from your code
* **Images**: Inline images
* **Files**: PDF documents
* **Code blocks**: Formatted code snippets
## Creating Messages for Reuse
Messages can be shared across prompts. Create standalone messages:
In your task, click **Messages** in the sidebar.
Click **Create Message**.
* **Name**: Descriptive name (e.g., "Standard System Prompt")
* **Role**: system, user, or assistant
* **Content**: The message content
When adding messages to prompts, select existing messages to reuse them.
### Benefits of Reuse
* Update once, affects all prompts using it
* Consistent instructions across prompts
* Easier maintenance
## Message Ordering
Drag and drop to reorder messages within a prompt:
* **Prompt: Product Help**
1. System Prompt (system)
2. Context Template (user)
3. Query Template (user)
The order determines the conversation flow sent to the LLM.
## Prompt Settings
Configure prompt behavior:
### Completion Config
Set default model settings:
* **Provider**: Anthropic, OpenAI, Google
* **Model**: claude-sonnet-4-20250514, gpt-4o, etc.
* **Max Tokens**: Maximum response length
* **Temperature**: Creativity (0-1)
### Tools
Attach tools for function calling:
1. Create a Schema with tool definitions
2. Attach to the prompt
3. Configure tool\_choice (auto, required, specific tool)
### Structured Output
For JSON responses:
1. Create a Schema defining the output structure
2. Attach as structured output
3. Responses will conform to the schema
## Deleting Content
Deletes are tracked per-branch and take effect on commit.
To delete:
1. Click the **...** menu on the item
2. Select **Delete**
3. Confirm the deletion
The item remains in other branches until you merge.
## Best Practices
Write clear system instructions first. They set the tone for everything.
Names appear in code and logs. Make them meaningful.
Use the Studio to test prompts before committing changes.
Share common instructions across prompts to ensure consistency.
## Next Steps
Add dynamic variables to messages
Version control your prompts
Use prompts in your code
Understand the data model
# Prompt Studio
Source: https://moxn.mintlify.app/webapp/studio
Test and iterate on prompts in real-time
The Prompt Studio is an interactive environment for testing and iterating on your prompts before using them in production.
![Prompt Studio interface]()
## Accessing the Studio
Access the Studio from multiple places:
1. **From the Dashboard**: Click **Studio** in the sidebar
2. **From a Task**: Click the **Studio** button in the task header
3. **From a Prompt**: Click **Open in Studio** on any prompt
## Studio Interface
### Header Controls
![Studio prompt selector]()
The header contains:
| Control | Description |
| ------------------- | -------------------------------------- |
| **Prompt Selector** | Choose which prompt to test |
| **Go to** | Navigate to the prompt's detail page |
| **Branch Selector** | Choose which branch to use |
| **Commit Selector** | Test specific versions |
| **Model Settings** | Configure provider, model, temperature |
| **Execute** | Run the prompt |
### Prompt Template Panel
The main editing area shows all messages in your prompt:
* **System messages** (blue background)
* **User messages** (gray background)
* **Assistant messages** (white background)
Edit messages directly in the Studio—changes auto-save to your working state.
### Variables Panel
On the right side, you'll see input variables:
* Each variable from your prompt's input schema appears here
* Fill in values to test different scenarios
* Supports text, objects, arrays, and media types
## Testing Prompts
### Basic Testing
Use the prompt selector dropdown to choose a prompt.
Enter values for each input variable.
Choose provider, model, temperature, and max tokens.
Run the prompt and see the response.
### Variable Types
| Type | How to Enter |
| ----------- | ----------------------------- |
| **String** | Plain text in the input field |
| **Number** | Numeric value |
| **Boolean** | Toggle switch |
| **Object** | JSON in code editor |
| **Array** | JSON array in code editor |
| **Image** | Upload or paste URL |
| **File** | Upload file |
### Viewing Results
After execution, you'll see:
* **Response content**: The LLM's output
* **Token usage**: Input and output token counts
* **Latency**: Time to generate response
* **Cost estimate**: Based on model pricing
Results are automatically logged for observability (visible in the Traces tab).
## Import from Logs
Replay previous executions by importing variable values from traces:
Open the import dialog.
Browse recent traces or search by metadata.
Variable values are populated from that execution.
Tweak values and execute again to compare.
This is useful for:
* Debugging production issues
* Testing edge cases from real data
* Building test cases from actual usage
## Test Cases
Create and manage test cases for systematic testing:
### Creating Test Cases
Fill in variable values for a test scenario.
Name your test case (e.g., "Password reset request").
Optionally add notes about expected output.
### Running Test Cases
1. Click **Test Cases** to see saved cases
2. Select one to load its variable values
3. Execute and compare results
4. Create observations for regression testing
## Multi-Turn Conversations
Test conversational prompts with multiple turns:
Run with user's first message.
Add the assistant response and a new user message.
Continue the conversation.
Useful for testing:
* Follow-up question handling
* Context retention
* Conversation flow
## Tool Calling
Test prompts with tools (function calling):
1. **Configure tools**: Attach schemas to your prompt
2. **Execute**: The model may return tool calls
3. **Provide results**: Enter mock tool results
4. **Continue**: See how the model uses the results
## Best Practices
Try empty inputs, very long inputs, and unusual characters.
Test the same prompt with different models to compare quality and cost.
Build a library of test cases for regression testing.
Always test prompt changes in Studio before committing.
Import from logs to test with production-like inputs.
## Next Steps
Create content to test
Define input variables
Analyze test results
Save tested changes
# Variables & Properties
Source: https://moxn.mintlify.app/webapp/variables
Add dynamic content to your prompts
Variables let you inject runtime data into your prompts. This guide covers how to add variables and define their types.
## Adding Variables
Variables are placeholders in messages that get replaced at runtime:
```
"Customer {{customer_name}} asks: {{query}}"
```
Becomes:
```
"Customer Alice asks: How do I reset my password?"
```
### Insert a Variable
Click on the message you want to edit.
In the editor, type `/variable` to open the variable menu.
* Select an existing property
* Or create a new one
Set the variable format:
* **Inline**: Embedded in text
* **Block**: On its own line
### Message Editor
The TipTap editor provides a rich editing experience:
![Message editor with TipTap]()
Features:
* Rich text formatting
* Variable insertion via `/variable` command
* Auto-save as you edit
* Visual distinction between text and variables
### Variable Syntax
Variables appear as `{{variable_name}}` in the editor:
```
System: You are a support agent for {{company_name}}.
User: Customer: {{customer_name}}
Query: {{query}}
Relevant documents:
{{search_results}}
```
## Properties
A **Property** defines a variable's type and metadata.
### Creating Properties
In your task, click **Properties** in the sidebar.
Click **Create Property**.
Configure the property:
* **Name**: Variable name (e.g., `query`)
* **Type**: string, number, object, array, etc.
* **Description**: What this variable represents
* **Required**: Whether it must be provided
### Property Types
| Type | Use Case | Example |
| --------- | --------------- | ------------------- |
| `string` | Text values | User queries, names |
| `integer` | Whole numbers | Counts, IDs |
| `number` | Decimals | Scores, prices |
| `boolean` | True/false | Flags |
| `object` | Structured data | User profiles |
| `array` | Lists | Search results |
### String Formats
Strings can have special formats:
```json theme={null}
{
"type": "string",
"format": "date"
}
```
Available formats:
* `date`: "2024-01-15"
* `date-time`: "2024-01-15T10:30:00Z"
* `email`: "[user@example.com](mailto:user@example.com)"
* `uri`: "[https://example.com](https://example.com)"
* `uuid`: "550e8400-e29b-..."
### Complex Types
#### Objects
Define nested structures:
```json theme={null}
{
"type": "object",
"properties": {
"id": {"type": "string"},
"name": {"type": "string"},
"email": {"type": "string", "format": "email"}
},
"required": ["id", "name"]
}
```
In the editor:
```
User profile: {{user}}
```
In code:
```python theme={null}
session_data = Input(
user=User(id="123", name="Alice", email="alice@example.com")
)
```
#### Arrays
Define lists:
```json theme={null}
{
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string"}
}
}
}
```
In the editor:
```
Search results:
{{documents}}
```
In code:
```python theme={null}
session_data = Input(
documents=[
Document(title="FAQ", content="..."),
Document(title="Guide", content="...")
]
)
```
## Input Schema Auto-Sync
When you add variables to messages, the prompt's **input schema** updates automatically.
### How It Works
```
1. You add {{query}} to a message
↓
2. System detects the variable
↓
3. Links to property "query"
↓
4. Adds "query" to input schema
↓
5. Codegen produces typed model with "query" field
```
### Viewing the Input Schema
1. Open your prompt
2. Click **Schema** tab
3. See all variables and their types
### Schemas Tab
View all schemas in the **Schemas** tab:
![Schemas tab overview]()
The tab shows:
* **Input Schemas**: Auto-generated from prompt variables
* **User-Defined Schemas**: Custom schemas for tools and structured output
## Variable Formatting
### Inline Variables
Embedded within text:
```
"Hello {{customer_name}}, how can I help?"
```
### Block Variables
On their own line, typically for larger content:
```
Context from search:
{{search_results}}
```
### Choosing Format
| Use Inline When | Use Block When |
| ------------------ | ------------------ |
| Short values | Long content |
| Part of a sentence | Standalone section |
| Names, IDs | JSON, documents |
## Referencing the Same Property
You can use the same property multiple times:
```
System: You help customers of {{company_name}}.
User: {{company_name}} customer {{customer_name}} asks:
{{query}}
```
The same value is substituted everywhere.
## Required vs Optional
Mark properties as required or optional:
```json theme={null}
{
"type": "object",
"properties": {
"query": {"type": "string"}, // Required
"context": {"type": "string"} // Optional
},
"required": ["query"]
}
```
In code:
```python theme={null}
class Input(RenderableModel):
query: str # Required
context: str | None = None # Optional
```
## Default Values
Properties can have default values:
```json theme={null}
{
"type": "string",
"default": "You are a helpful assistant."
}
```
If not provided at runtime, the default is used.
## Best Practices
`customer_query` is better than `q` or `input1`.
Descriptions help teammates understand what each variable is for.
Use specific types (integer, date) rather than just string when possible.
Use objects to group related fields (user info, search context).
Each prompt should only have the variables it actually uses.
## Next Steps
Save your changes
Generate typed models
Deep dive into the variable system
Use variables at runtime
# Branching & Committing
Source: https://moxn.mintlify.app/webapp/versioning
Version control for your prompts
Moxn uses Git-like version control for prompts. This guide covers branching, committing, and managing versions in the web app.
## Understanding Branches
A **branch** is an isolated workspace for changes:
* **main**: The default branch (like Git's main/master)
* **Feature branches**: For experimentation and development
```mermaid theme={null}
graph TD
subgraph main [main]
m1[Commit 1: Initial prompts]
m2[Commit 2: Improved system prompt]
m3[Working: Current uncommitted changes]
m1 --> m2 --> m3
end
subgraph feature [feature-improvements]
f1["(inherits from main)"]
f2[Working: Experimental changes]
f1 --> f2
end
```
## Creating Branches
Click the branch dropdown in the header.
Select **Create new branch**.
Use descriptive names:
* `feature-better-prompts`
* `experiment-new-tone`
* `fix-edge-case`
Click **Create** to create and switch to the new branch.
### Branch Selector
![Branch selector dropdown]()
The branch dropdown shows:
* Current branch (with checkmark)
* All available branches
* Option to create new branch
### Git Actions Menu
![Git Actions menu]()
Click **Git Actions** to:
* Create a new branch
* Create a merge request
* View branch settings
### Branch Naming
Good branch names:
* `feature-multilingual-support`
* `experiment-shorter-responses`
* `fix-formatting-issue`
Avoid:
* `test`
* `my-branch`
* `changes`
## Switching Branches
To switch between branches:
1. Click the branch dropdown
2. Select the branch you want
3. Your view updates to show that branch's content
Uncommitted changes stay on their branch when you switch.
## Making Changes
### Working State
Changes you make are saved as **working state**:
* Automatically saved as you edit
* Visible only on your current branch
* Not visible to others until you commit
### What Counts as a Change
* Editing message content
* Adding or removing messages
* Modifying prompt settings
* Adding or removing properties
* Changing schemas
## Committing Changes
When you're ready to save a snapshot:
Check the **Changes** panel to see what's modified.
Click the **Commit** button.
Describe what you changed:
* "Improved system prompt for clarity"
* "Added context variable for search results"
* "Fixed typo in greeting"
Click **Commit** to create the snapshot.
### Commit Messages
Write clear commit messages:
Good:
* "Add multilingual greeting support"
* "Reduce system prompt length for faster responses"
* "Add search\_results variable for RAG"
Avoid:
* "Updates"
* "Fixed stuff"
* "."
## Viewing History
See the commit history for your branch:
Click the commit dropdown (shows current commit or "Working").
See all commits with:
* Commit message
* Timestamp
* Changes summary
Click to view that commit's snapshot.
### Commit History
![Commit history selector]()
The commit selector shows:
* **Latest**: Current working state
* **Commit history**: Past snapshots with messages and timestamps
* Click any commit to view that version
### Viewing Past Versions
When viewing a past commit:
* Content is **read-only**
* You're seeing exactly what was committed
* Make changes by switching back to "Working"
## Comparing Versions
Compare different versions:
1. Select the first commit/version
2. Click **Compare**
3. Select the second commit/version
4. See a diff view showing changes
## Rolling Back
To revert to a previous version:
Select the commit you want to restore.
Click **Restore to this version**.
This creates new working state matching that commit.
Commit the restored state with a message like "Revert to commit abc123".
## Merging Branches
To merge changes from one branch to another:
Switch to the branch you want to merge INTO (e.g., main).
Click **Merge** and select the source branch.
See what will be merged.
If there are conflicts, choose which version to keep.
Click **Merge** to apply the changes.
## Deployment Workflow
### Development
1. Create feature branch
2. Make changes
3. Test in Studio
4. Commit when satisfied
5. Merge to main
### Production
1. Get latest commit ID from main
2. Update code to use that commit ID
3. Deploy
In code:
```python theme={null}
# Production: pinned to commit
session = await client.create_prompt_session(
prompt_id="...",
commit_id="abc123def456" # From your deployment config
)
```
## Branch Protection
Consider protecting your main branch to prevent accidental changes.
Protection options:
* Require reviews before merging
* Require commit messages
* Lock branch from direct edits
## Best Practices
Small, frequent commits are easier to track and revert.
Don't experiment directly on main. Create branches.
Future you will thank present you.
Use the Studio to verify prompts work as expected.
Never use branch names in production. Always use commit IDs.
## Next Steps
Deep dive into versioning
Access versions in code
Create new content
Track which versions are used