docs

Agent

The Agent class is the core of the SDK. Create an agent, call .run() with a message, and get a result with the output, token count, and cost.

Creating an agent

Python
from ninetrix import Agent

agent = Agent(
    name="researcher",
    provider="anthropic",                # anthropic, openai, google, mistral, groq
    model="claude-sonnet-4-6",
    temperature=0.2,
    max_tokens=8192,
    max_turns=20,
    instructions="You are a research assistant. Be thorough and cite sources.",
    tools=[my_tool_1, my_tool_2],        # @Tool-decorated functions
)

Running an agent

Python
# Synchronous (event-loop-safe — uses thread pool internally)
result = agent.run("What are the latest developments in quantum computing?")

# Async
result = await agent.arun("What are the latest developments in quantum computing?")

# Access the result
print(result.output)          # The agent's final text response
print(result.tokens_used)     # Total tokens (input + output)
print(result.input_tokens)    # Input tokens only
print(result.output_tokens)   # Output tokens only
print(result.cost_usd)        # Estimated cost in USD
print(result.steps)           # Number of tool-use turns
print(result.thread_id)       # Thread ID for resumption
print(result.history)         # Full message history
print(result.error)           # None on success, error message on failure

Streaming

Use .stream() for real-time output — tokens arrive as they're generated:

Python
async for event in agent.stream("Explain quantum entanglement"):
    match event.type:
        case "token":
            print(event.content, end="", flush=True)
        case "tool_start":
            print(f"\n[Calling {event.tool_name}...]")
        case "tool_end":
            print(f"[{event.tool_name} done]")
        case "done":
            print(f"\nTotal cost: ${event.cost_usd:.4f}")

Multi-turn conversations

Pass a thread_id to resume a previous conversation. The agent loads its history from the checkpointer and continues where it left off.

Python
# First message
result = agent.run("Find me 5 papers on transformer architecture", thread_id="research-session")

# Continue the conversation (history is restored)
result = agent.run("Now summarize the third paper in detail", thread_id="research-session")
Requires a checkpointer
Multi-turn resumption requires a Checkpointer (e.g. PostgresCheckpointer). Without one, thread_id is set but history is not persisted across process restarts. See Persistence.

Introspection

Python
# Inspect agent config (no API call)
info = agent.info()
print(info.name, info.model, info.tools)

# Validate configuration before running
issues = agent.validate()
for issue in issues:
    print(f"{issue.level}: {issue.message}")

# Dry run — simulate without calling the LLM
dry = agent.dry_run("test message")
print(dry.system_prompt, dry.tool_schemas)
On this page