JamJet

pip install jamjet

from jamjet import task, tool

@tool
async def web_search(query: str) -> str:
    return f"Search results for: {query}"

@task(model="claude-haiku-4-5-20251001", tools=[web_search])
async def research(question: str) -> str:
    """You are a research assistant. Search first, then summarize clearly."""

result = await research("What is JamJet?")
print(result)

pip install jamjet
jamjet init my-first-agent
cd my-first-agent
jamjet dev

jamjet run workflow.yaml --input '{"query": "What is JamJet?"}'

# workflow.yaml
workflow:
  id: hello-agent
  version: 0.1.0
  state_schema:
    query: str
    answer: str
  start: think

nodes:
  think:
    type: model
    model: claude-haiku-4-5-20251001
    prompt: "Answer clearly and concisely: {{ state.query }}"
    output_key: answer
    next: end

  end:
    type: end

jamjet validate workflow.yaml
jamjet run workflow.yaml --input '{"query": "What is JamJet?"}'

from jamjet import task, tool

@tool
async def web_search(query: str) -> str:
    return f"Search results for: {query}"

@task(model="claude-haiku-4-5-20251001", tools=[web_search])
async def research(question: str) -> str:
    """You are a research assistant. Search first, then summarize clearly."""

result = await research("What is JamJet?")

from jamjet import Agent, tool

@tool
async def web_search(query: str) -> str:
    return f"Search results for: {query}"

agent = Agent(
    "researcher",
    model="claude-haiku-4-5-20251001",
    tools=[web_search],
    instructions="You are a research assistant. Search first, then summarize.",
)

result = await agent.run("What is JamJet?")
print(result)

from jamjet import Workflow, tool
from pydantic import BaseModel

@tool
async def web_search(query: str) -> str:
    return f"Search results for: {query}"

workflow = Workflow("research")

@workflow.state
class State(BaseModel):
    query: str
    answer: str | None = None

@workflow.step
async def search(state: State) -> State:
    result = await web_search(query=state.query)
    return state.model_copy(update={"answer": result})

nodes:
  search:
    type: tool
    server: brave-search        # configured in jamjet.toml
    tool: web_search
    arguments:
      query: "{{ state.query }}"
      count: 10
    output_key: results
    next: summarize

nodes:
  delegate:
    type: a2a_task
    agent_url: "https://agents.example.com/research-agent"
    input:
      query: "{{ state.query }}"
    output_key: research
    next: end

nodes:
  check:
    type: eval
    scorers:
      - type: llm_judge
        rubric: "Is the answer accurate and complete?"
        min_score: 4
    on_fail: retry_with_feedback   # injects feedback into next model call
    max_retries: 2
    next: end

from jamjet.coordinator import DefaultCoordinatorStrategy

strategy = DefaultCoordinatorStrategy(registry=my_registry)

# Discover agents by skill, score them, route to the best fit
candidates, _ = await strategy.discover(
    task="Analyze quarterly revenue data",
    required_skills=["data-analysis", "finance"],
    trust_domain="internal",
)
rankings, spread = await strategy.score(task, candidates, weights={})
decision = await strategy.decide(task, rankings, threshold=0.1)
# decision.selected_uri → "jamjet://org/finance-analyst"

from jamjet.agent_tool import agent_tool

# Sync: quick, stateless
classifier = agent_tool(agent="jamjet://org/classifier", mode="sync",
                        description="Classifies documents by topic")

# Streaming: long-running with early termination on budget
researcher = agent_tool(agent="jamjet://org/researcher", mode="streaming",
                        description="Deep research with progress", budget={"max_cost_usd": 2.00})

# Conversational: multi-turn iterative refinement
reviewer = agent_tool(agent="jamjet://org/reviewer", mode="conversational",
                      description="Peer review with feedback", max_turns=5)

from jamjet.workflow.graph import WorkflowGraph

graph = WorkflowGraph("pipeline")
graph.add_agent_tool("process", agent="auto", mode="sync", output_key="result")
# ↑ "auto" expands at compile time into: Coordinator → AgentTool
ir = graph.compile()
# IR now has 2 nodes: _coordinator_process → process

Is it a single task?
  → Single Agent

Does order matter?
  → Sequential Pipeline

Can tasks run independently?
  → Parallel Fan-Out

Does output need quality checks?
  → Loop & Critic

Do you need to pick the best agent at runtime?
  → Coordinator

Does one agent need to invoke another?
  → Agent-as-Tool