Add JEV decision routing examples

docs/agents/reference/agent-schema.json

@@ -40,6 +40,9 @@
       "properties": {
         "name": { "type": "string", "pattern": "^[a-zA-Z_][a-zA-Z0-9_-]*$" },
         "model": { "type": ["string", "null"] },
+        "kind": { "const": "decision" },
+        "provider": { "type": "string" },
+        "questions": { "type": "object", "additionalProperties": { "type": "object" } },
         "baseUrl": { "type": ["string", "null"] },
         "strategy": { "type": ["string", "null"] },
         "maxTurns": { "type": ["integer", "null"], "minimum": 0 },

examples/agentic_workflows/README.md

@@ -2,10 +2,13 @@
 
 AI/LLM workflow examples using Conductor's built-in system tasks (`LLM_CHAT_COMPLETE`, `LLM_INDEX_TEXT`, `LLM_SEARCH_INDEX`, MCP tools) combined with Python workers.
 
-All examples use **inline ChatMessage objects** for system prompts -- no named prompt templates or AIOrchestrator required. They work with OSS Conductor with AI/LLM support.
+Chat examples use **inline ChatMessage objects** for system prompts -- no named prompt templates or AIOrchestrator required. They work with OSS Conductor with AI/LLM support.
 
 ## Prerequisites
 
+The decision routing example requires server-side decision-evaluator support and
+Decision credentials. No chat model or Python worker is needed. The other examples require:
+
 - Conductor server with AI/LLM support running (e.g., `http://localhost:7001/api`)
 - LLM provider named `openai` configured with a valid API key
 - `export CONDUCTOR_SERVER_URL=http://localhost:7001/api`
@@ -14,6 +17,7 @@ All examples use **inline ChatMessage objects** for system prompts -- no named p
 
 | Example | Description | Interactive? | Pattern |
 |---------|-------------|:------------:|---------|
+| [ai_decision_routing.py](ai_decision_routing.py) | Route requests and return the selected branch's result | No | Decision-backed SwitchTask + InlineTask |
 | [llm_chat.py](llm_chat.py) | Automated multi-turn science Q&A between two LLMs | No | LoopTask + LLM_CHAT_COMPLETE + worker for history |
 | [llm_chat_human_in_loop.py](llm_chat_human_in_loop.py) | Interactive chat with WAIT task pauses for user input | Yes | LoopTask + WaitTask + LLM_CHAT_COMPLETE |
 | [multiagent_chat.py](multiagent_chat.py) | Multi-agent debate with moderator routing between panelists | No | LoopTask + SwitchTask + SetVariableTask + JavaScript routing |
@@ -23,6 +27,9 @@ All examples use **inline ChatMessage objects** for system prompts -- no named p
 ## Quick Start
 
 ```bash
+# Decision request routing with server-managed credentials
+CONDUCTOR_SERVER_URL=http://localhost:8080/api python -m examples.agentic_workflows.ai_decision_routing
+
 # Automated multi-turn chat (no interaction needed)
 python examples/agentic_workflows/llm_chat.py
 
@@ -38,6 +45,19 @@ python examples/agentic_workflows/function_calling_example.py
 
 ## Key Patterns
 
+### AI decision routing
+
+Set `SwitchTask.evaluator_type = "decision"`. Its input parameters contain
+`model`, optional `provider`, `state`, and exactly one choice question in
+`questions`; its `expression` names that question. The choice keys must match the
+task's `decisionCases` keys. The server performs inference, writes the full
+response and `selectedCase` to the SWITCH output, and runs the selected branch.
+Evaluation runs synchronously during task mapping. Provider failures fail the
+workflow without task retries.
+
+Agent definitions with a decision `ToolDef` in `router=` compile the selector and
+selected child workflow into this same decision-backed SWITCH shape automatically.
+
 ### Passing dynamic messages to LLM_CHAT_COMPLETE
 
 When passing a workflow reference as `messages`, set it via `input_parameters` AFTER construction:

examples/agentic_workflows/ai_decision_routing.py

@@ -0,0 +1,100 @@
+"""Route a request with one decision-backed SWITCH and an inline branch.
+
+The server evaluates the SWITCH synchronously and selects its branch. Provider
+failures fail the workflow without task retries. No Python worker is needed.
+Run: CONDUCTOR_SERVER_URL=http://localhost:8080/api python -m examples.agentic_workflows.ai_decision_routing
+"""
+
+import argparse
+import json
+import time
+
+from conductor.client.configuration.configuration import Configuration
+from conductor.client.orkes_clients import OrkesClients
+from conductor.client.workflow.conductor_workflow import ConductorWorkflow
+from conductor.client.workflow.task.inline import InlineTask
+from conductor.client.workflow.task.switch_task import EvaluatorType, SwitchTask
+
+
+def create_workflow(executor, model="jev-1.13", provider=None) -> ConductorWorkflow:
+    workflow = ConductorWorkflow(executor=executor, name="ai_decision_routing", version=1)
+    decision = SwitchTask(
+        task_ref_name="route_request",
+        case_expression="route",
+    )
+    decision.evaluator_type = EvaluatorType.DECISION.value
+    decision.input_parameters = {
+        "model": model,
+        **({"provider": provider} if provider is not None else {}),
+        "state": workflow.input("request"),
+        "questions": {
+            "route": {
+                "type": "choice",
+                "instructions": "Choose the team best suited to handle this request.",
+                "choices": {
+                    "billing": "Payments, invoices, refunds, or subscriptions.",
+                    "technical": "Errors, outages, or product troubleshooting.",
+                },
+            }
+        },
+    }
+    billing = InlineTask(
+        "handle_billing",
+        script='({team: "billing", nextAction: "review_invoice", '
+        'message: "Check invoice line items and payment records.", request: $.request})',
+        bindings={"request": workflow.input("request")},
+    )
+    technical = InlineTask(
+        "handle_technical",
+        script='({team: "technical", nextAction: "collect_diagnostics", '
+        'message: "Collect error logs and steps to reproduce.", request: $.request})',
+        bindings={"request": workflow.input("request")},
+    )
+    decision.switch_case("billing", [billing])
+    decision.switch_case("technical", [technical])
+    result = InlineTask(
+        "selected_result",
+        script="$.billing || $.technical",
+        bindings={
+            "billing": billing.output("result"),
+            "technical": technical.output("result"),
+        },
+    )
+    workflow >> decision >> result
+    workflow.output_parameters(
+        {
+            "selectedCase": decision.output("selectedCase"),
+            "answers": decision.output("answers"),
+            "usage": decision.output("usage"),
+            "cost": decision.output("usage.cost"),
+            "latencyMs": decision.output("latencyMs"),
+            "result": result.output("result"),
+        }
+    )
+    return workflow
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--request", default="I was charged twice on my latest invoice.")
+    parser.add_argument("--model", default="jev-1.13")
+    parser.add_argument("--provider")
+    args = parser.parse_args()
+    clients = OrkesClients(configuration=Configuration())
+    workflow = create_workflow(clients.get_workflow_executor(), args.model, args.provider)
+    workflow.register(overwrite=True)
+    workflow_id = workflow.start_workflow_with_input({"request": args.request})
+    print(f"Workflow: {workflow_id}")
+    client = clients.get_workflow_client()
+    while True:
+        result = client.get_workflow(workflow_id=workflow_id, include_tasks=False)
+        if result.is_completed():
+            break
+        time.sleep(1)
+    if result.status != "COMPLETED":
+        raise SystemExit(f"{result.status}: {result.reason_for_incompletion}")
+    print(json.dumps(result.output, indent=2))
+
+
+if __name__ == "__main__":
+    main()

examples/agents/README.md

@@ -26,3 +26,15 @@ the Conductor CLI manages the server with `conductor server start`.
 Framework-specific examples are in [ADK](adk/README.md),
 [LangGraph](langgraph/README.md), and [OpenAI Agents SDK](openai/README.md).
 Review tool side effects before using real credentials.
+
+Decision: [decision_router.py](decision_router.py) uses a decision `ToolDef` as a
+router selector, with normal chat agents as executable children. Deployment
+compiles it into one decision-backed `SWITCH`; the tool is not deployed as a
+standalone agent. It compiles by default; pass `--run` for inference.
+
+`decision_tool.py` shows the other agent pattern: a normal chat agent calls
+Decision inference as a server-side tool. The server maps the tool call to a
+branchless decision-backed `SWITCH`; no tool worker is needed.
+
+For direct use in a traditional workflow, see
+[`ai_decision_routing.py`](../agentic_workflows/ai_decision_routing.py).

examples/agents/decision_router.py

@@ -0,0 +1,86 @@
+"""Route to one of two chat agents with a decision tool.
+
+When a decision ToolDef is supplied as ``router=``, deployment compiles it into
+one decision-backed SWITCH. It is a selector configuration, not a standalone
+agent, and needs no Python worker. Pass --run for inference.
+"""
+
+import argparse
+import json
+import os
+
+from conductor.ai.agents import Agent, AgentRuntime, Strategy, ToolDef
+from conductor.client.configuration.configuration import Configuration
+
+PROMPT = "The customer reports a duplicate charge on the latest invoice."
+CHAT_MODEL = os.getenv("CONDUCTOR_AGENT_LLM_MODEL", "openai/gpt-4o-mini")
+
+
+def support_router():
+    billing = Agent(
+        name="billing",
+        model=CHAT_MODEL,
+        instructions="Help the customer with billing and invoice questions.",
+    )
+    technical = Agent(
+        name="technical",
+        model=CHAT_MODEL,
+        instructions="Help the customer troubleshoot product issues.",
+    )
+    selector = ToolDef(
+        name="support_selector",
+        description="Choose the support agent that should handle the request.",
+        tool_type="decision",
+        config={
+            "model": "jev-1.13",
+            **(
+                {"provider": os.environ["CONDUCTOR_DECISION_PROVIDER"]}
+                if "CONDUCTOR_DECISION_PROVIDER" in os.environ
+                else {}
+            ),
+            "questions": {
+                "agent": {
+                    "type": "choice",
+                    "instructions": "Choose the right support team.",
+                    "choices": {
+                        "billing": "Payments and invoices",
+                        "technical": "Product errors and troubleshooting",
+                    },
+                }
+            },
+        },
+    )
+    return Agent(
+        name="decision_support_router",
+        strategy=Strategy.ROUTER,
+        router=selector,
+        agents=[billing, technical],
+        max_turns=1,
+        synthesize=False,
+    )
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--run", action="store_true")
+    args = parser.parse_args()
+    config = Configuration(
+        server_api_url=os.environ.get("CONDUCTOR_SERVER_URL", "http://localhost:8080/api")
+    )
+    with AgentRuntime(config) as runtime:
+        agent = support_router()
+        if not args.run:
+            print(json.dumps(runtime.plan(agent), indent=2))
+            return
+
+        runtime.deploy(agent)
+        handle = runtime.start(agent, PROMPT)
+        print("Execution:", handle.execution_id)
+        result = handle.join(timeout=120)
+        if not result.is_success:
+            raise RuntimeError(f"{result.status}: {result.error}")
+        print(json.dumps(result.output, indent=2))
+
+
+if __name__ == "__main__":
+    main()

examples/agents/decision_tool.py

@@ -0,0 +1,66 @@
+"""Use Decision inference as a tool inside a normal chat agent.
+
+The server executes the tool as a branchless decision-backed SWITCH, so no Python
+worker is needed.
+"""
+
+import os
+
+from conductor.ai.agents import Agent, AgentRuntime, ToolDef
+from conductor.client.configuration.configuration import Configuration
+
+decision_config = {
+    "model": "jev-1.13",
+    **(
+        {"provider": os.environ["CONDUCTOR_DECISION_PROVIDER"]}
+        if "CONDUCTOR_DECISION_PROVIDER" in os.environ
+        else {}
+    ),
+    "questions": {
+        "department": {
+            "type": "choice",
+            "instructions": "Choose the correct support department.",
+            "choices": {
+                "billing": "Payments and invoices",
+                "technical": "Product errors and troubleshooting",
+            },
+        }
+    },
+}
+
+classify_request = ToolDef(
+    name="classify_request",
+    description="Classify the customer request before answering.",
+    tool_type="decision",
+    input_schema={
+        "type": "object",
+        "properties": {"state": {"type": "string"}},
+        "required": ["state"],
+    },
+    config=decision_config,
+)
+
+support_agent = Agent(
+    name="decision_tool_support",
+    model=os.getenv("CONDUCTOR_AGENT_LLM_MODEL", "openai/gpt-4o-mini"),
+    instructions="Call classify_request before answering the customer.",
+    tools=[classify_request],
+)
+
+
+def main():
+    config = Configuration(
+        server_api_url=os.getenv("CONDUCTOR_SERVER_URL", "http://localhost:8080/api")
+    )
+    with AgentRuntime(config) as runtime:
+        result = runtime.run(
+            support_agent,
+            "I was charged twice on my latest invoice.",
+        )
+        if not result.is_success:
+            raise RuntimeError(result.error)
+        print(result.output)
+
+
+if __name__ == "__main__":
+    main()

src/conductor/ai/agents/agent.py

@@ -15,6 +15,7 @@
 from typing import Any, Callable, Dict, List, Optional, Union
 
 from conductor.ai.agents.claude_code import ClaudeCode
+from conductor.ai.agents.tool import ToolDef
 
 _VALID_NAME_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_-]*$")
 
@@ -496,8 +497,8 @@ class Agent:
             ``"handoff"``).  Valid values: ``handoff``, ``sequential``,
             ``parallel``, ``router``, ``round_robin``, ``random``, ``swarm``,
             ``manual``.
-        router: For ``strategy="router"``, an :class:`Agent` or callable that
-            selects which sub-agent runs each turn.
+        router: For ``strategy="router"``, an :class:`Agent`, callable, or
+            decision :class:`ToolDef` that selects which sub-agent runs each turn.
         output_type: A Pydantic model or dataclass for structured output.
         guardrails: List of :class:`Guardrail` instances for input/output validation.
         memory: Optional :class:`ConversationMemory` for session management.
@@ -555,7 +556,7 @@ def __init__(
         tools: Optional[List[Any]] = None,
         agents: Optional[List[Any]] = None,
         strategy: Union[str, Strategy] = Strategy.HANDOFF,
-        router: Optional[Union["Agent", Callable[..., Any]]] = None,
+        router: Optional[Union["Agent", Callable[..., Any], "ToolDef"]] = None,
         output_type: Optional[type] = None,
         guardrails: Optional[List[Any]] = None,
         memory: Optional[Any] = None,
@@ -913,11 +914,14 @@ def is_claude_code(self) -> bool:
     def external(self) -> bool:
         """``True`` if this agent references an external workflow (no local definition).
 
-        An agent with no ``model`` is treated as external — the server
-        produces a ``SubWorkflowTask`` referencing the workflow by name
-        instead of compiling the agent inline.
+        An agent with no model references an existing workflow, except a
+        ROUTER with a decision tool, which the server compiles inline.
         """
-        return not self.model
+        return not self.model and not (
+            self.strategy == Strategy.ROUTER
+            and isinstance(self.router, ToolDef)
+            and self.router.tool_type == "decision"
+        )
 
     # ── Instance-method resolution ──────────────────────────────────────