Escribe flujos de trabajo de JamJet en Python usando decoradores y el constructor de workflows.

SDK de Python

El SDK de Python de JamJet te permite escribir flujos de trabajo en Python en lugar de YAML. Ambos se compilan a la misma IR y se ejecutan en el mismo runtime de Rust.

Instalación

pip install jamjet

API de Decoradores

La API de decoradores es la forma más concisa de escribir flujos de trabajo. Decora una función con @node y JamJet infiere el tipo de nodo a partir de la firma de la función y el tipo de retorno.

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}

Ejecútalo:

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())

Constructor de flujos de trabajo

Para mayor control, usa la API de constructor:

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()
)

Acceso al estado

State es un objeto tipo dict tipado. Accede a las claves con state["key"] o state.key:

@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}

consejo: Los nodos devuelven parches de estado, no el estado completo. Solo necesitas devolver las claves que quieres actualizar. Las claves existentes se conservan.

Llamadas al modelo

Usa self.model() dentro de cualquier nodo para llamar a un LLM:

@node
async def think(self, state: State) -> State:
    response = await self.model(
        model="claude-sonnet-4-6",
        prompt=f"Answer: {state['query']}",
        system="You are concise and accurate.",
        temperature=0.3,
        max_tokens=512,
    )

    # response.text — texto completo
    # response.usage.input_tokens
    # response.usage.output_tokens
    # response.model

    return {"answer": response.text}

Llamadas a herramientas (MCP)

Usa self.tool() para llamar a una herramienta desde un servidor MCP conectado:

@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}

Llamadas HTTP

@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()}

Ramificación

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 {}  # el nodo de ramificación lee el estado existente — no requiere salida

Ejecución paralela

from jamjet.workflows import ParallelNode

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

Políticas de reintento

from jamjet.workflows import RetryPolicy

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

Ejecutar flujos de trabajo

from jamjet import JamJetClient

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

# Ejecutar y esperar la finalización

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

# Ejecutar sin esperar — recibe un ID de ejecución inmediatamente

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

# Consultar el estado

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

# Transmitir eventos en tiempo real

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

Anotaciones de tipos

El SDK incluye stubs de tipos completos. En modo estricto:

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)

Configuración

from jamjet import JamJetClient, JamJetConfig

client = JamJetClient(config=JamJetConfig(
    base_url="http://localhost:7700",
    api_key="YOUR_API_KEY",  # para hosted/producción
    timeout_ms=30_000,
    default_model="claude-haiku-4-5-20251001",
))

O mediante variables de entorno:

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

SDK de Python

On this page