Python SDK for Norns.
pip install norns-sdkThe SDK has two parts. Norns is the worker — it connects to the server, registers your agent, and sits in a loop handling tasks. NornsClient is for sending messages and reading results from application code (your Slack bot, web backend, CLI, etc).
import os
from norns import Norns, Agent, tool
@tool
def search_docs(query: str) -> str:
"""Search product documentation."""
return db.vector_search(query)
@tool(side_effect=True)
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email to a customer."""
smtp.send(to=to, subject=subject, body=body)
return f"Email sent to {to}"
agent = Agent(
name="support-bot",
model="claude-sonnet-4-20250514",
system_prompt="You are a customer support agent. Look up docs and help customers.",
tools=[search_docs, send_email],
mode="conversation",
on_failure="retry_last_step",
)
norns = Norns("http://localhost:4000", api_key=os.environ["NORNS_API_KEY"])
norns.run(agent) # LLM API keys read from env (ANTHROPIC_API_KEY, OPENAI_API_KEY, etc.)norns.run() connects via WebSocket, registers the agent and tools, then blocks forever handling llm_task and tool_task dispatches. LLM calls go through LiteLLM, so any supported provider works. Norns never sees your API keys — your worker makes all external calls.
import os
from norns import NornsClient
client = NornsClient("http://localhost:4000", api_key=os.environ["NORNS_API_KEY"])
# Fire-and-forget
run = client.send_message("support-bot", "Where's my order?")
# run.run_id, run.status == "accepted"
# Wait for completion
result = client.send_message("support-bot", "Where's my order?", wait=True, timeout=30)
print(result.output)
# Multi-turn with a conversation key
result = client.send_message("support-bot", "And the tracking number?",
conversation_key="slack:U01ABC", wait=True)
# Inspect a run
run = client.get_run(42)
events = client.get_events(42)
# Stream events as they happen
for event in client.stream("support-bot", "Research quantum computing"):
if event.type == "completed":
print(event.data.get("output", "")[:80])
breakThe @tool decorator infers JSON Schema from type hints. The docstring becomes the tool description the LLM sees.
@tool
def lookup_customer(email: str) -> str:
"""Look up a customer by email."""
customer = db.query("SELECT * FROM customers WHERE email = ?", email)
return f"Found: {customer['name']} ({customer['plan']})"Mark side-effecting tools so Norns can enforce idempotency on replay:
@tool(side_effect=True)
def charge_card(customer_id: str, amount: float) -> str:
"""Charge a credit card."""
result = stripe.charges.create(customer=customer_id, amount=int(amount * 100))
return f"Charged ${amount}: {result['id']}"Async handlers work too:
@tool
async def fetch_page(url: str) -> str:
"""Fetch a web page."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.text()agent = Agent(
name="my-agent",
model="claude-sonnet-4-20250514",
system_prompt="You are helpful.",
tools=[search, send_email],
mode="conversation", # "task" or "conversation"
checkpoint_policy="on_tool_call", # "every_step", "on_tool_call", "manual"
context_window=20,
max_steps=50,
on_failure="retry_last_step", # "stop" or "retry_last_step"
)MIT