Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.go-mizu.dev/llms.txt

Use this file to discover all available pages before exploring further.

The Python SDK generator creates modern Python client libraries from your contract definitions. The generated code features full type hints, dataclass models, and both synchronous and asynchronous clients powered by httpx.

​
Key Features

  • Modern Python: Python 3.8+ with full type hints
  • Sync and Async: Both Client (sync) and AsyncClient (async) clients
  • httpx Powered: Industry-standard HTTP client with automatic retries
  • Dataclass Models: Clean, typed data structures
  • uv-Ready: Includes pyproject.toml for modern Python packaging
  • SSE Streaming: First-class Server-Sent Events support

​
Quick Start

​
Step 1: Define Your Contract

# api.yaml
name: TodoAPI
description: Todo list API

defaults:
  base_url: http://localhost:8080

resources:
  - name: todos
    description: Manage todo items
    methods:
      - name: create
        input: CreateInput
        output: Todo
        http:
          method: POST
          path: /todos
      - name: list
        output: ListOutput
        http:
          method: GET
          path: /todos
      - name: get
        input: GetInput
        output: Todo
        http:
          method: GET
          path: /todos/{id}
      - name: delete
        input: DeleteInput
        http:
          method: DELETE
          path: /todos/{id}

types:
  - name: Todo
    kind: struct
    fields:
      - name: id
        type: string
      - name: title
        type: string
      - name: done
        type: bool
  - name: CreateInput
    kind: struct
    fields:
      - name: title
        type: string
  - name: GetInput
    kind: struct
    fields:
      - name: id
        type: string
  - name: DeleteInput
    kind: struct
    fields:
      - name: id
        type: string
  - name: ListOutput
    kind: struct
    fields:
      - name: items
        type: "[]Todo"
      - name: total
        type: int

​
Step 2: Generate the SDK

mizu contract gen api.yaml --client --lang python --output ./sdk/python --package todoclient --version 1.0.0

​
Step 3: Install and Use

# Install the generated SDK
cd sdk/python
uv pip install -e .

# Or with pip
pip install -e .

from todoclient import Client

# Create a client
client = Client(base_url="http://localhost:8080")

# Create a todo
todo = client.todos.create(title="Learn Mizu Python SDK")
print(f"Created: {todo.id}")

# List all todos
result = client.todos.list()
print(f"Total: {result.total} todos")

# Get a specific todo
todo = client.todos.get(id=todo.id)
print(f"Got: {todo.id} - {todo.title}")

# Delete
client.todos.delete(id=todo.id)
print("Deleted successfully")

​
Generated Code Structure

sdk/python/
├── pyproject.toml
└── src/
    └── todoclient/
        ├── __init__.py
        ├── _client.py
        ├── _types.py
        ├── _resource.py
        └── _streaming.py

​
pyproject.toml

[project]
name = "todoclient"
version = "1.0.0"
requires-python = ">=3.8"
dependencies = ["httpx>=0.24.0"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

​
init.py

from ._client import Client, AsyncClient

__all__ = ["Client", "AsyncClient"]

​
Client Configuration

​
Creating a Client

from todoclient import Client

# Basic client
client = Client(base_url="http://localhost:8080")

# With authentication
client = Client(
    api_key="your-api-key",
    base_url="http://localhost:8080",
)

# With all options
client = Client(
    api_key="your-api-key",
    base_url="http://localhost:8080",
    timeout=30.0,
    max_retries=3,
    default_headers={"X-Custom-Header": "value"},
)

​
Configuration Options

OptionTypeDefaultDescription
api_keystr | NoneNoneAuthentication token
base_urlstr | NoneService defaultAPI base URL
timeoutfloat | NoneNone (no timeout)Request timeout in seconds
max_retriesint2Maximum retry attempts
default_headersDict[str, str] | NoneNoneHeaders for all requests

​
Modifying Client Options

Create a new client with modified options: 
# Original client
client = Client(api_key="key-1", base_url="http://localhost:8080")

# New client with different API key
other_client = client.with_options(api_key="key-2")

# New client with additional headers
debug_client = client.with_options(
    default_headers={"X-Debug": "true"}
)

​
Closing the Client

Always close the client when done: 
client = Client(base_url="http://localhost:8080")
try:
    # Use client...
    pass
finally:
    client.close()
Or use a context manager pattern: 
from contextlib import closing

with closing(Client(base_url="http://localhost:8080")) as client:
    todo = client.todos.create(title="Task")

​
Async Client

The SDK includes a fully async client for use with asyncio: 
import asyncio
from todoclient import AsyncClient

async def main():
    client = AsyncClient(base_url="http://localhost:8080")

    try:
        # Create a todo
        todo = await client.todos.create(title="Async task")
        print(f"Created: {todo.id}")

        # List todos
        result = await client.todos.list()
        for t in result.items:
            print(f"- {t.title}")

    finally:
        await client.close()

asyncio.run(main())

​
Async Context Manager

async def main():
    client = AsyncClient(base_url="http://localhost:8080")
    try:
        todos = await client.todos.list()
    finally:
        await client.aclose()  # or await client.close()

​
Type System

​
Type Mapping Reference

Contract TypePython Type
stringstr
boolbool
int, int8-int64int
uint, uint8-uint64int
float32, float64float
time.Timedatetime
json.RawMessageobject
anyobject
[]TList[T]
map[string]TDict[str, T]

​
Dataclass Types

Contract struct types generate Python dataclasses: 
from dataclasses import dataclass
from typing import Optional
from datetime import datetime

@dataclass
class Todo:
    id: str
    title: str
    done: bool
    created_at: datetime

​
Optional and Nullable Fields

Contract DefinitionPython Type
Required fieldT
optional: trueOptional[T]
nullable: trueOptional[T]
Both optional and nullableOptional[T]
Example: 
@dataclass
class UpdateInput:
    title: str                    # required
    done: Optional[bool] = None   # optional

​
Enum Fields

Enum fields use literal type hints in the docstring: 
@dataclass
class Task:
    # status: One of "pending", "active", "completed"
    status: str

​
List and Dict Types

from typing import List, Dict

# Contract: kind: slice, elem: Todo
TodoList = List[Todo]

# Contract: kind: map, elem: string
Metadata = Dict[str, str]

​
Resources and Methods

​
Resource Pattern

Each contract resource becomes a property on the client: 
client.todos      # TodosResource
client.users      # UsersResource

​
Method Signatures

Methods use Python naming conventions (snake_case):
Contract MethodPython Signature
Createcreate(**kwargs) -> Todo
Listlist() -> ListOutput
Getget(id: str) -> Todo
Deletedelete(id: str) -> None

​
Calling Methods

Methods accept keyword arguments matching the input type fields: 
# If input type is CreateInput { title: string }
todo = client.todos.create(title="My task")

# If input type is UpdateInput { title: string, done: bool (optional) }
todo = client.todos.update(id="123", title="Updated", done=True)

​
Streaming (SSE)

For methods with streaming support, the SDK provides iterator-based consumption:

​
Sync Streaming

# Stream events
for event in client.responses.stream(model="gpt-4", input="Hello"):
    if event.type == "text_delta":
        print(event.text, end="", flush=True)
    elif event.type == "completed":
        print("\n--- Done ---")

​
Async Streaming

async for event in client.responses.stream(model="gpt-4", input="Hello"):
    if event.type == "text_delta":
        print(event.text, end="", flush=True)

​
Collecting All Events

# Collect all events into a list
events = list(client.responses.stream(model="gpt-4", input="Hello"))

# Or async
events = [event async for event in client.responses.stream(...)]

​
Error Handling

​
Error Types

The SDK defines three error types: 
class SDKError(Exception):
    """Base exception for all SDK errors."""
    pass

class APIConnectionError(SDKError):
    """Raised when unable to connect to the API."""
    pass

class APIStatusError(SDKError):
    """Raised when the API returns an error status code."""
    status_code: int
    body: Any

​
Handling Errors

from todoclient import Client
from todoclient._client import APIStatusError, APIConnectionError

client = Client(base_url="http://localhost:8080")

try:
    todo = client.todos.get(id="nonexistent")
except APIStatusError as e:
    if e.status_code == 404:
        print("Todo not found")
    elif e.status_code == 401:
        print("Unauthorized")
    elif e.status_code == 400:
        print(f"Bad request: {e}")
    else:
        print(f"API error {e.status_code}: {e}")
except APIConnectionError as e:
    print(f"Connection failed: {e}")

​
Automatic Retries

The client automatically retries failed requests (except for 4xx errors): 
# Configure max retries
client = Client(
    base_url="http://localhost:8080",
    max_retries=5,  # Retry up to 5 times
)
Retries are attempted for:
  • Network errors
  • 5xx server errors
  • Timeout errors

​
Advanced Usage

​
Custom Headers

# Default headers for all requests
client = Client(
    base_url="http://localhost:8080",
    default_headers={
        "X-Request-ID": "abc-123",
        "X-API-Version": "2024-01",
    },
)

​
Authentication Modes

The SDK supports different authentication modes based on your service configuration: 
# Bearer token (default)
client = Client(api_key="your-token")
# Sends: Authorization: Bearer your-token

# Basic auth (if service configured with auth: basic)
client = Client(api_key="base64-encoded-credentials")
# Sends: Authorization: Basic base64-encoded-credentials

# No auth (if service configured with auth: none)
client = Client()
# Sends no Authorization header

​
Timeout Configuration

# Set timeout in seconds
client = Client(
    base_url="http://localhost:8080",
    timeout=30.0,  # 30 second timeout
)

# No timeout (wait indefinitely)
client = Client(
    base_url="http://localhost:8080",
    timeout=None,
)

​
Using with Existing httpx Client

The generated SDK uses httpx internally. For advanced use cases, you can access the underlying client: 
# The SDK creates its own httpx client
client = Client(base_url="http://localhost:8080")
# client._http is the httpx.Client instance

​
Complete Example

​
Server

// main.go
package main

import (
    "github.com/go-mizu/mizu"
    contract "github.com/go-mizu/mizu/contract/v2"
    "github.com/go-mizu/mizu/contract/v2/transport/rest"

    "yourapp/todo"
)

func main() {
    impl := todo.NewService()
    svc := contract.Register[todo.API](impl,
        contract.WithDefaultResource("todos"),
        contract.WithDefaults(&contract.Defaults{
            BaseURL: "http://localhost:8080",
        }),
    )

    app := mizu.New()
    rest.Mount(app.Router, svc)
    app.Listen(":8080")
}

​
Client Usage

#!/usr/bin/env python3
"""Example usage of the generated Todo client."""

from todoclient import Client

def main():
    # Create client
    client = Client(base_url="http://localhost:8080")

    try:
        # Create a todo
        todo = client.todos.create(title="Buy groceries")
        print(f"Created: {todo.id}")

        # List all todos
        result = client.todos.list()
        print(f"\nAll todos ({result.total} total):")
        for t in result.items:
            status = "x" if t.done else " "
            print(f"  [{status}] {t.id}: {t.title}")

        # Get a specific todo
        fetched = client.todos.get(id=todo.id)
        print(f"\nFetched: {fetched.title}")

        # Delete the todo
        client.todos.delete(id=todo.id)
        print(f"\nDeleted: {todo.id}")

    finally:
        client.close()

if __name__ == "__main__":
    main()

​
Async Example

#!/usr/bin/env python3
"""Async example usage of the generated Todo client."""

import asyncio
from todoclient import AsyncClient

async def main():
    client = AsyncClient(base_url="http://localhost:8080")

    try:
        # Create multiple todos concurrently
        tasks = [
            client.todos.create(title="Task 1"),
            client.todos.create(title="Task 2"),
            client.todos.create(title="Task 3"),
        ]
        todos = await asyncio.gather(*tasks)

        for todo in todos:
            print(f"Created: {todo.id} - {todo.title}")

        # List all
        result = await client.todos.list()
        print(f"\nTotal: {result.total} todos")

    finally:
        await client.close()

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

​
Installation and Distribution

​
Installing Locally

# Using uv (recommended)
cd sdk/python
uv pip install -e .

# Using pip
cd sdk/python
pip install -e .

​
Publishing to PyPI

cd sdk/python

# Build the package
uv build
# or: python -m build

# Upload to PyPI
uv publish
# or: twine upload dist/*

​
Installing from Git

pip install git+https://github.com/yourorg/yourrepo.git#subdirectory=sdk/python

​
See Also