mokra-sdk 1.0.0

Mokra SDK - World tests for AI agents - intercept and mock HTTP requests across services

mokra

Mokra SDK - World tests for AI agents - intercept and mock HTTP requests across services

Mokra intercepts HTTP requests from your AI agent and redirects them to mock endpoints. Configure which services to intercept, and Mokra handles the rest.

Installation

pip install mokra

Quick Start

import mokra

# Configure connection to MockServer
mokra.configure(
    base_url="https://api.mokra.ai/api",
    api_key="your-api-key"
)

# Create a world with your services
world = mokra.mockworld("my test", "stripe", "shopify")

# Run your agent
def agent_code(services):
    import requests
    # Your agent code here - HTTP calls are automatically intercepted
    requests.post(
        "https://api.stripe.com/v1/charges",
        json={"amount": 1000, "currency": "usd"}
    )

world.run(agent_code)

# See what happened
world.observe()

# Assert outcomes
world.assert_(
    "a charge was created",
    "no errors occurred"
)

How It Works

┌─────────────────────────────────────────────────────────────┐
│                     Your Agent Code                         │
│  ┌─────────────────────────────────────────────────────┐   │
│  │           world.run(lambda services:               │   │
│  │               requests.post("https://api.stripe...│   │
│  │           )                                        │   │
│  └─────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                     Mokra Interceptor                       │
│                                                             │
│   https://api.stripe.com/v1/charges                         │
│                     ↓                                       │
│   https://api.mokra.ai/api/mock/stripe/v1/charges           │
│                                                             │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│              Mock Server (api.mokra.ai)                     │
│                                                             │
│    Records observations, returns mock responses,            │
│    evaluates assertions, manages state                      │
└─────────────────────────────────────────────────────────────┘

Configuration

Direct Configuration

mokra.configure(
    base_url="https://api.mokra.ai/api",
    api_key="your-api-key"
)

Environment Variables

export MOKRA_BASE_URL="https://api.mokra.ai/api"
export MOKRA_API_KEY="your-api-key"

Auth Styles

# Default: X-API-Key header (for mokra.ai)
mokra.configure(
    api_key="your-api-key"
)
# Sends: X-API-Key: your-api-key

# Bearer token auth
mokra.configure(
    api_key="your-token",
    auth_style="bearer"
)
# Sends: Authorization: Bearer your-token

# Custom header
mokra.configure(
    api_key="your-key",
    auth_style="custom",
    auth_header="X-Custom-Auth"
)
# Sends: X-Custom-Auth: your-key

API Reference

mokra.mockworld(name, *services)

Create a new world with specified services.

# String shorthand
world = mokra.mockworld("my test", "stripe", "shopify", "loop")

# Attach existing mock server
world = mokra.mockworld("my test",
    "stripe",
    {"name": "shopify", "server_id": "existing-ms-id"}
)

world.run(func) / with world.run():

Run your agent. All HTTP calls made during execution are intercepted and recorded by MockServer.

# Function style
def agent_code(services):
    # Your agent code
    requests.post("https://api.stripe.com/v1/charges", ...)

world.run(agent_code)

# Context manager style (recommended for pytest)
with world.run():
    requests.post("https://api.stripe.com/v1/charges", ...)

world.assert_(*assertions)

Assert conditions using natural language.

results = world.assert_(
    "customer charged once",
    "no refund issued"
)

world.observe()

Display observations recorded by MockServer.

world.observe()

world.seed(func)

Pre-populate state before running the agent.

def setup(s):
    s.shopify.orders.create({"id": "order-123", "total": 99.99})

world.seed(setup)

world.state(mock_server_id=None)

Get current state of a mock server.

state = world.state()
state.total                    # => 5
state.resource_types           # => ["products", "orders"]
state["products"].count        # => 3
state["products"].records      # => [StateRecord, ...]
state["products"].find_by_id("prod-123")

Service Mappings

Mokra uses YAML files to map hosts to service slugs. The default mappings for mokra.ai are included.

Custom Mappings

from mokra import ServiceMapper

# Add a single mapping
ServiceMapper.add_mapping("api.myservice.com", "myservice")

# Load mappings from a YAML file
ServiceMapper.load_mappings("path/to/custom.yml")

# Load and replace all mappings
ServiceMapper.load_mappings("path/to/custom.yml", replace=True)

Custom YAML Format

# custom_mappings.yml
api.myservice.com: myservice
api.another.com: another
*.internal.mycompany.com: internal

Supporting Other Mock Servers

Create a YAML file in the mappings directory named after your provider:

# my_provider.yml
api.stripe.com: stripe
api.custom.com: custom

Then load it:

from mokra import ServiceMapper

ServiceMapper.load_provider("my_provider")

Pytest Integration

Mokra works seamlessly with pytest:

# conftest.py
import pytest
import mokra
import os

@pytest.fixture(autouse=True)
def configure_mokra():
    mokra.configure(api_key=os.environ["MOKRA_API_KEY"])

# test_refund.py
from mokra import mockworld

def test_refund_service():
    world = mockworld(name="Refund test", services_list=["stripe"])

    with world.run():
        # Your code that calls Stripe API
        RefundService.process(payment_id="pi_123", amount=5000)

    world.assert_("a refund was created")
    world.assert_("refund amount is $50")

Development

pip install -e ".[dev]"
pytest

License

MIT