Python SDK

Das JamJet Python SDK ermöglicht es dir, Workflows in Python statt in YAML zu schreiben. Beide kompilieren zur gleichen IR und laufen auf derselben Rust-Runtime.

Installation

pip install jamjet

Decorator API

Die Decorator API ist die prägnanteste Art, Workflows zu schreiben. Dekoriere eine Funktion mit @node und JamJet leitet den Node-Typ aus der Funktionssignatur und dem Rückgabetyp ab.

from jamjet import workflow, node, State

@workflow(id="hello-agent", version="0.1.0")
class HelloWorkflow:
    @node(start=True)
    async def think(self, state: State) -> State:
        response = await self.model(
            model="claude-haiku-4-5-20251001",
            prompt=f"Answer clearly: {state['query']}",
        )
        return {"answer": response.text}

Ausführen:

import asyncio
from jamjet import JamJetClient

async def main():
    client = JamJetClient()  # connects to http://localhost:7700
    result = await client.run(
        HelloWorkflow,
        input={"query": "What is JamJet?"}
    )
    print(result.state["answer"])

asyncio.run(main())

Workflow-Builder

Für mehr Kontrolle nutze die Builder API:

from jamjet.workflows import WorkflowBuilder, ModelNode, ToolNode, BranchNode

wf = (
    WorkflowBuilder("research-agent", version="0.2.0")
    .state_schema(query=str, results=list, answer=str)
    .add_node(
        ToolNode("search")
        .server("brave-search")
        .tool("web_search")
        .arguments({"query": "{{ state.query }}", "count": 5})
        .output_key("results")
        .next("draft")
    )
    .add_node(
        ModelNode("draft")
        .model("claude-sonnet-4-6")
        .prompt("""
            Search results: {{ state.results | join('\\n') }}

            Answer: {{ state.query }}
        """)
        .output_key("answer")
        .next("end")
    )
    .start("search")
    .build()
)

State-Zugriff

State ist ein typisiertes dict-ähnliches Objekt. Greife auf Keys mit state["key"] oder state.key zu:

@node
async def process(self, state: State) -> State:
    query = state["query"]          # raises KeyError if missing
    context = state.get("context")  # returns None if missing

    # Return a partial state patch — only keys you include are updated
    return {"answer": "...", "confidence": 0.95}

Tipp: Nodes geben State-Patches zurück, nicht den vollständigen State. Du musst nur die Keys zurückgeben, die du aktualisieren möchtest. Bestehende Keys bleiben erhalten.

Modellaufrufe

Verwenden Sie self.model() innerhalb eines beliebigen Knotens, um ein LLM aufzurufen:

@node
async def think(self, state: State) -> State:
    response = await self.model(
        model="claude-sonnet-4-6",
        prompt=f"Antwort: {state['query']}",
        system="Du bist prägnant und präzise.",
        temperature=0.3,
        max_tokens=512,
    )

    # response.text — vollständiger Text
    # response.usage.input_tokens
    # response.usage.output_tokens
    # response.model

    return {"answer": response.text}

Tool-Aufrufe (MCP)

Verwenden Sie self.tool(), um ein Tool von einem verbundenen MCP-Server aufzurufen:

@node
async def search(self, state: State) -> State:
    result = await self.tool(
        server="brave-search",
        tool="web_search",
        arguments={"query": state["query"], "count": 5},
    )
    return {"results": result.content}

HTTP-Aufrufe

@node
async def fetch(self, state: State) -> State:
    result = await self.http(
        method="GET",
        url=f"https://api.example.com/items/{state['item_id']}",
        headers={"Authorization": f"Bearer {self.env('API_KEY')}"},
    )
    return {"raw": result.json()}

Verzweigung

from jamjet import node, branch

@node
@branch(
    conditions=[
        ("state['confidence'] >= 0.9", "done"),
        ("state['confidence'] >= 0.5", "refine"),
    ],
    default="escalate",
)
async def route(self, state: State) -> State:
    return {}  # Verzweigungsknoten liest vorhandenen State — keine Ausgabe erforderlich

Parallele Ausführung

from jamjet.workflows import ParallelNode

.add_node(
    ParallelNode("gather")
    .branches(["search", "fetch-docs", "check-cache"])
    .join("synthesize")
)

Retry-Richtlinien

from jamjet.workflows import RetryPolicy

ModelNode("think")
    .model("claude-haiku-4-5-20251001")
    .prompt("...")
    .retry(RetryPolicy(
        max_attempts=3,
        backoff="exponential",
        delay_ms=500,
    ))

Workflows ausführen

from jamjet import JamJetClient

client = JamJetClient(base_url="http://localhost:7700")

# Ausführen und auf Abschluss warten

result = await client.run(wf, input={"query": "..."})
print(result.state)
print(result.execution_id)
print(result.duration_ms)

# Fire and Forget — sofort eine Execution-ID zurückerhalten

exec_id = await client.submit(wf, input={"query": "..."})

# Status abfragen

status = await client.get_execution(exec_id)
print(status.status)  # running | completed | failed

# Ereignisse in Echtzeit streamen

async for event in client.stream(wf, input={"query": "..."}):
    print(event.type, event.node_id)

Typannotationen

Das SDK enthält vollständige Type Stubs. Im Strict Mode:

from jamjet import State, NodeResult
from typing import TypedDict

class MyState(TypedDict):
    query: str
    answer: str
    confidence: float

@node(start=True)
async def think(self, state: MyState) -> NodeResult[MyState]:
    ...
    return NodeResult(answer="...", confidence=0.9)

Konfiguration

from jamjet import JamJetClient, JamJetConfig

client = JamJetClient(config=JamJetConfig(
    base_url="http://localhost:7700",
    api_key="YOUR_API_KEY",  # für Hosted/Production
    timeout_ms=30_000,
    default_model="claude-haiku-4-5-20251001",
))

Oder über Umgebungsvariablen:

export JAMJET_URL=http://localhost:7700
export JAMJET_API_KEY=...

Python SDK

On this page