nineth 0.5.57

model sdk built by the 9th ditrict at tooig

nineth

nineth is the Python SDK for the 1984 model API, built by the 9th District at Tooig.

---

Install

pip install nineth
export NINETH_API_KEY="your-api-key"

---

How it works

Every request goes through client.model.request(...).

• Pass a task. Get a response.
• Set stream=True to receive text as it arrives, word by word.
• Per request, you can opt into persistent sessions, built-in services, included services, a base-provider override, caller policy text, audio input, outbound messaging transports, JSON output, compute totals, and verbose telemetry.
• The server still runs the worker loop and manages the actual task state.

---

Models

Name	Description
1984-m3-0317	Most capable. Best for research and complex tasks.
1984-m2-preview	Fast and powerful. Good for most tasks.
1984-m2-light	Lightweight, quick general tasks.
1984-m1-unified	High-throughput unified model.
1984-m0-brute	Compact efficient model.
1984-m0-sm	Smallest model, fastest responses.

Set a default at client creation or pass model= per call.

---

Cookbook

1 — Get a response

The simplest case. Ask something, get the answer.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request("Give me a tight BTC market brief.")
    print(response["final_response"])

response is a plain dict. The text is always in response["final_response"].

---

2 — Stream the response live

Set stream=True to print text as it arrives.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    for event in client.model.request("Summarise crude oil today.", stream=True):
        if event["type"] == "model_delta":
            print(event["data"]["text"], end="", flush=True)

The last event in the stream is type: result and contains the full final_response alongside iterations.

If continuous=True, the server keeps the worker alive after a non-terminal idle turn, but it now waits internally for interrupts or due alarms instead of spending another model turn just to emit a monitoring prompt.

Built-in service progress is streamed as {"type": "service_call", "data": {"service_name": "...", "client_managed": false}}.

• If default_service=False, built-in service progress events are hidden.
• If default_service=[...], only built-in service names from that allowlist are surfaced.
• If messaging={...} is set, the SDK automatically enables the related messaging services so their progress and service_response events can surface.
• For email, messaging.email.email and messaging.email.name define the sender mailbox identity (from), not the eventual recipient, and messaging.email.instruction configures how that inbox should answer inbound mail.
• Messaging-related service_response events include the sanitized raw result payload so callers can track email send, webhook, or delivery state.
• Client-managed services included by the caller are not executed by the server. Their callback stream stays raw via model_raw_delta and client_service_call so your code can parse and execute them client-side.

---

3 — Choose a different model per request

from nineth import NinethClient

with NinethClient() as client:
    response = client.model.request(
        "What happened with Nvidia earnings?",
        model="1984-m2-light",
    )
    print(response["final_response"])

---

4 — Control reasoning depth

Use reasoning to hint at how deeply the model should think before answering. Valid values: "low", "medium", "high". Leave it out to use the model default.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Analyse the macro impact of a Fed rate pause.",
        reasoning="high",
    )
    print(response["final_response"])

---

5 — Show the model's reasoning

Set show_reasoning=True to include the model's internal chain-of-thought. This is off by default.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Walk me through whether gold is trending or ranging.",
        reasoning="medium",
        show_reasoning=True,
    )
    for block in response.get("thinking", []):
        print("[thinking]", block)
    print(response["final_response"])

---

6 — Limit how many turns the model takes

max_iterations controls how many model turns the server runs. The default is 10. Most tasks finish in 1–3 turns.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Give me a one-paragraph ETH brief.",
        max_iterations=2,
    )
    print(response["final_response"])

continuous is separate from max_iterations.

• continuous=False: the request finishes on a non-terminal idle turn.
• continuous=True: the server keeps the worker open and waits internally for interrupts or alarms without spending another model turn while idle.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Watch for the next alarm and continue only when it fires.",
        default_service=["set_alarm"],
        continuous=True,
        max_iterations=20,
    )
    print(response["final_response"])

---

7 — Async usage

import asyncio
from nineth import AsyncNinethClient

async def main():
    async with AsyncNinethClient(default_model="1984-m3-0317") as client:
        response = await client.model.request(
            "Summarise macro risk factors this week.",
        )
        print(response["final_response"])

asyncio.run(main())

Async streaming works the same way:

import asyncio
from nineth import AsyncNinethClient

async def main():
    async with AsyncNinethClient(default_model="1984-m3-0317") as client:
        async for event in await client.model.request(
            "Research BTC ETF flows.", stream=True
        ):
            if event["type"] == "model_delta":
                print(event["data"]["text"], end="", flush=True)

asyncio.run(main())

---

8 — Health check

No API key needed. Use this to verify the endpoint is reachable.

from nineth import NinethClient

with NinethClient() as client:
    print(client.health())
# {'status': 'ok', 'timestamp': '2026-04-04T00:00:00+00:00'}

---

9 — Provider routing

SDK requests use the base-system provider path by default. Set base_system=False only when you explicitly want to fall back to the runtime default provider selection.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0421") as client:
    response = client.model.request(
        "Summarise today's macro tape.",
        base_system=False,
    )
    print(response["final_response"])

base_system is True by default.

---

10 — Add caller policy text

Use policy= to add caller instructions on top of the SDK/API runtime prompt. This does not remove the runtime done or service-call rules.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Summarise today's macro tape as compact JSON.",
        policy="Return a concise JSON object with keys summary and risks.",
        response_format="json",
    )
    print(response["final_response"])

---

11 — Send audio input

Pass audio= as base64 strings or dicts with data, optional mime_type, and optional filename. The server stores and transcribes the audio, then injects the transcript into the task context.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Summarise the attached voice memo.",
        audio=[{"data": "BASE64_AUDIO_HERE", "mime_type": "audio/wav", "filename": "memo.wav"}],
    )
    print(response["final_response"])

---

12 — Request JSON output and compute totals

Use response_format="json" to ask for JSON output. When the final response is valid JSON, the SDK parses it into final_response and preserves the original text in raw_response.

Use compute=True to surface the total token count in compute. This is the total of prompt plus completion tokens.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Return the answer as JSON with keys answer and confidence.",
        response_format="json",
        compute=True,
    )
    print(response["final_response"])
    print(response["raw_response"])
    print(response["compute"])

---

13 — Configure mailbox messaging

Use messaging= to attach request-scoped email or Telegram transport defaults. Blank strings or default keep the server-side defaults for email sender details. For email, email and name are the sender mailbox identity (from), and instruction configures how that inbox should respond when inbound mail is processed later.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Email the latest risk summary to ops and tell me when it is sent.",
        messaging={
            "email": {
                "email": "[e-mail]",
                "name": "Billing Bot",
                "instruction": "Ask for the invoice number before confirming payment.",
            }
        },
        stream=False,
    )
    print(response["final_response"])
    print(response.get("service_responses", []))

The email status surface for that mailbox returns mailbox_config and recent_logs, including received payload summaries, model responses, and sanitized Resend state.

You can also attach ownership-aware templates for multi-mailbox workflows. This is useful when the caller identity determines both how inbound mail is interpreted and how outbound mail should be rendered.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Process review mail and respond using the owner's template.",
        messaging={
            "email": {
                "email": "[e-mail]",
                "name": "Review Desk",
                "instruction": "Handle inbound review requests.",
                "templates": [
                    {
                        "type": "inbound",
                        "name": "review_request",
                        "required": ["ticket_id", "review_url"],
                        "url": "https://api.example.com/email/review-context",
                    },
                    {
                        "type": "outbound",
                        "name": "review_reply",
                        "required": ["decision", "reviewer_name"],
                        "recipients": ["[e-mail]"],
                    },
                ],
            }
        },
    )

Template behavior:

• messaging.email.templates accepts inbound and outbound template definitions.
• Inbound templates require url; the server posts the inbound email context there before the model runs.
• Outbound templates require recipients; the model must provide every field listed in required.
• Ownership verification normalizes addresses and checks that inbound mail belongs to the configured mailbox before template processing runs.

Inbound callback payloads include fields such as from, to, subject, text, and message_id. The callback must return a JSON object of template variables, for example:

{
    "ticket_id": "rvw-123",
    "review_url": "https://portal.example.com/reviews/rvw-123"
}

When the model answers with send_email or send_reply, it can target the owner template explicitly:

{
    "send_reply": {
        "owner_template_name": "review_reply",
        "owner_template_params": {
            "decision": "approved",
            "reviewer_name": "A. Chen"
        }
    }
}

The server validates the required template parameters before dispatch and passes them through to the email provider as template variables.

Parameter handling is split cleanly:

• messaging.email.templates is request-scoped config sent by the SDK as part of the normal messaging payload.
• For inbound templates, the server calls the configured url and expects a JSON object of variables.
• For outbound templates, the model emits owner_template_name plus owner_template_params inside send_email or send_reply.
• The SDK does not pre-validate owner_template_params; the server validates that every field named in required is present before dispatch.
• Messaging config auto-enables send_email and send_reply in the SDK payload, so you do not need to list them manually in default_service.

Hosted inbound webhook notes (Modal):

• The inbound endpoint (EMAIL_WEBHOOK_URL) receives Resend webhook events and validates Svix headers (Svix-Id, Svix-Timestamp, Svix-Signature).
• For type == "email.received", the webhook requires data.email_id and queues async processing.
• email_id is not the mailbox address. It is the Resend receiving message identifier from the webhook payload.
• The mailbox address (for example [e-mail]) is used as the inbox identity (to) and ownership target, not as the email_id fetch key.
• If email_id is missing, the endpoint returns {"success": false, "message": "Missing email_id"}.
• If email_id is present, the endpoint may return queued success even if downstream fetch later fails; inspect email status logs for post-queue outcomes.

Template optionality:

• If no inbound templates are configured, inbound processing runs normally with no ownership-template callback section.
• If inbound templates are configured, the processor attempts ownership verification and URL parameter fetch, then injects template context into the model input.
• Outbound template rendering remains optional and only applies when the model emits owner_template_name plus owner_template_params.

Live Modal webhook contract test:

RUN_LIVE_WEBHOOK_TESTS=1 ./.venv/bin/pytest tests/test_email_webhook_live_contract.py -q

This test signs real webhook payloads with RESEND_WEBHOOK_SECRET and validates hosted responses for:

• non-email.received events
• email.received without email_id
• email.received with a placeholder email_id (queue contract only)

For streaming requests, messaging-related transport updates arrive as service_response events. The final plain-text answer still remains in the normal result event.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    for event in client.model.request(
        "Send the status email and stream delivery updates.",
        stream=True,
        messaging={"email": {"email": "[e-mail]"}},
    ):
        if event["type"] == "service_response":
            print(event["data"])

If you want stdout to reflect everything the SDK yields, handle all visible event types yourself. The SDK yields events immediately; flush=True is the caller-side print behavior, not an internal SDK buffer-flush feature.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    for event in client.model.request(
        "Draft the reply, send it, and surface the delivery details.",
        stream=True,
        messaging={
            "email": {
                "email": "[e-mail]",
                "name": "Review Desk",
                "templates": [
                    {
                        "type": "outbound",
                        "name": "review_reply",
                        "required": ["decision", "reviewer_name"],
                        "recipients": ["[e-mail]"],
                    }
                ],
            }
        },
    ):
        event_type = event["type"]

        if event_type == "model_delta":
            print(event["data"].get("text", ""), end="", flush=True)
        elif event_type == "service_call":
            print(f"\n[SERVICE_CALL] {event['data']}", flush=True)
        elif event_type == "service_response":
            print(f"\n[SERVICE_RESPONSE] {event['data']}", flush=True)
        elif event_type == "result":
            print("\n[RESULT]", flush=True)
            print(event["data"], flush=True)

Typical streaming shape for an email transport run:

{"type": "accepted", "data": {...}}
{"type": "model_delta", "data": {"text": "I drafted the reply..."}}
{"type": "model_delta", "data": {"text": "\n> Sending reply\n", "progress": True, "synthetic": True}}
{"type": "service_call", "data": {"service_name": "send_reply", "client_managed": False}}
{"type": "service_response", "data": {
    "service_name": "send_reply",
    "success": True,
    "summary": {"status": "sent"},
    "result": {"message_id": "email_123", "delivery": {"status": "sent"}}
}}
{"type": "result", "data": {
    "final_response": "Reply sent.",
    "iterations": 1,
    "usage": {...},
    "thinking": [],
    "service_calls": [...],
    "service_responses": [...]
}}

---

14 — Persist and reuse a session

Set cache=True to persist the task session and receive a reusable session_id. Pass that session_id back into later requests to continue the same task memory.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    first = client.model.request(
        "Start a running research notebook for crude oil.",
        cache=True,
    )
    session_id = first["session_id"]

    second = client.model.request(
        "Continue from the last note and add today's macro drivers.",
        cache=True,
        session_id=session_id,
    )
    print(second["final_response"])

If cache=False, the task is treated as one-shot and no session_id is exposed.

---

15 — Opt into built-in services

Built-in services are off by default.

• default_service=False: no built-in services
• default_service=True: all built-in services
• default_service=[...]: only the named built-in services

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Research OPEC headlines and summarise the impact.",
        default_service=["search_news", "search_web"],
    )
    print(response["final_response"])

---

16 — Use one include_service surface for custom services

include_service is the only custom-service surface you need.

• If you pass a local schema.py path or a directory containing schema.py, the SDK loads it locally, exposes its schema to the model, executes the emitted service call in your process, and resumes the run through the callback protocol.
• Buffered requests still auto-execute local callback services in the caller process.
• Streaming requests do not auto-execute local callback services. They pause with an awaiting_client_services event so your code can decide how to execute the local work and when to resume.
• If you pass a shorthand string such as a service-manager name, service name, or local service directory name, the SDK scans local schema.py files and prefers matches under services/ directories.
• If you pass a service manager class or instance that comes from a schema module, the SDK resolves that module automatically.
• If you pass an unresolved path string, the SDK passes it through to the API server as a server-hosted schema.py path.
• If a local service name collides with an existing server-side service name, the request fails early with a clear error instead of running the wrong service.

This keeps the public surface small while still supporting both local and server-managed services.

Local callback example:

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Use my local weather service and compare it with today's oil move.",
        default_service=["search_web"],
        include_service=["./services/weather/schema.py"],
    )
    print(response["final_response"])

Shorthand discovery example:

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Use my local weather service.",
        include_service=["WeatherServiceManager"],
    )
    print(response["final_response"])

Direct manager reference example:

from nineth import NinethClient
from myapp.schema import WeatherServiceManager

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Use my local weather service.",
        include_service=[WeatherServiceManager],
    )
    print(response["final_response"])

Manual streaming callback example:

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    pending_process_id = None

    for event in client.model.request(
        "What is the weather like right now in sf?",
        include_service=["./regulator/schema.py"],
        stream=True,
    ):
        if event["type"] == "model_delta":
            print(event["data"].get("text", ""), end="", flush=True)
        elif event["type"] == "awaiting_client_services":
            pending_process_id = event["process_id"]
            pending_calls = event["data"]["pending_client_calls"]

    if pending_process_id:
        resume = client.model.request(
            "Resume after local weather call.",
            stream=True,
            session_id=pending_process_id,
            client_service_results=[
                {
                    "call_id": pending_calls[0]["call_id"],
                    "service_name": "get_weather",
                    "success": True,
                    "result": {"location": "San Francisco", "forecast": "sunny"},
                }
            ],
        )
        for event in resume:
            if event["type"] == "model_delta":
                print(event["data"].get("text", ""), end="", flush=True)

Server-hosted path example:

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Use the custom weather service and compare it with market risk sentiment.",
        default_service=["search_web"],
        include_service=["/srv/app/services/weather/schema.py"],
    )
    print(response["final_response"])

Minimal local schema.py shape:

from typing import Any, Dict


class WeatherServiceManager:
    def get_weather(self, location: str) -> Dict[str, Any]:
        return {"location": location, "forecast": "sunny"}


weather_services = {
    "get_weather": {
        "name": "get_weather",
        "description": "Return a tiny weather forecast.",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string"}
            },
            "required": ["location"],
            "additionalProperties": False,
        },
    }
}


weather_implementations = {
    "get_weather": lambda manager: manager.get_weather,
}

17 — Enable verbose telemetry

Set verbose=True to keep the worker's structured telemetry in the buffered response or streamed events. The legacy debug=True alias still works.

from nineth import NinethClient

with NinethClient(default_model="1984-m3-0317") as client:
    response = client.model.request(
        "Research market risk and show the internal trace.",
        default_service=["search_news"],
        verbose=True,
    )
    print(response.get("events", []))

---

Response shape

Buffered (stream=False)

{
    "final_response": "Bitcoin is trading near...",
    "iterations": 2,
    "compute": 500,
    "usage": {"prompt_tokens": 412, "completion_tokens": 88, "total_tokens": 500},
    "thinking": [],          # only populated when show_reasoning=True
    "service_calls": [...],
    "service_responses": [...],
    "events": [...],
}

Only final_response and iterations are guaranteed to be present on every response. When response_format="json" and the final response is valid JSON, final_response is the parsed object and raw_response preserves the original string. compute is only included when requested and matches the provider total token count. When messaging is active, relevant service_responses entries may include a raw result payload with transport identifiers or delivery state.

For template-backed email sends, the buffered service_responses shape may include the same transport payload while the actual template variables remain part of the executed service call parameters:

{
    "final_response": "Reply sent.",
    "iterations": 1,
    "service_calls": [
        {
            "service_name": "send_reply",
            "params": {
                "owner_template_name": "review_reply",
                "owner_template_params": {
                    "decision": "approved",
                    "reviewer_name": "A. Chen"
                }
            }
        }
    ],
    "service_responses": [
        {
            "service_name": "send_reply",
            "success": True,
            "summary": {"status": "sent"},
            "result": {
                "message_id": "email_123",
                "delivery": {"status": "sent"}
            }
        }
    ]
}

Streaming (stream=True)

Each loop iteration yields a dict:

# Text arriving live
{"type": "model_delta", "data": {"text": "Bitcoin is trading..."}}

# service calls the model made
{"type": "service_call",     "data": {"service_name": "search_web", "params": {...}}}
{"type": "service_response", "data": {"service_name": "search_web", "success": True, "summary": {...}}}

# messaging-enabled service updates can also include raw transport state
{"type": "service_response", "data": {"service_name": "send_email", "success": True, "summary": {...}, "result": {"message_id": "..."}}}

# Final summary — always the last event
{"type": "result", "data": {"final_response": "...", "iterations": 2, "compute": 500}}

---

Error handling

from nineth import NinethClient, NinethAPIError

with NinethClient(default_model="1984-m3-0317") as client:
    try:
        response = client.model.request("Analyse ETH.")
    except NinethAPIError as exc:
        print("API error:", exc)
    except ValueError as exc:
        print("Configuration error:", exc)

SDK-visible error messages scrub transport or provider product names to tooig and strip URLs before raising NinethAPIError.

---

Authentication

Set NINETH_API_KEY in your environment or pass api_key= to the client constructor. That key can be any one key registered by the server's NINETH_API_KEYS registry. The health check endpoint does not require a key.