mistralai-vibe-protocol

SDK for the Task protocol — composable, observable, durable agentic orchestration.

Overview

mistralai-vibe-protocol is built around the Task, a single abstraction that unifies tools, agents, and workflows under one protocol. Every Task shares the same interface: input, output, streaming history updates, a terminal result event, and recursive subtask composition.

• One interface for everything: Tools, LLM agents, workflows, and sandboxed code are all Tasks
• Recursive composability: Tasks compose into observable trees at any depth
• Streaming observability: History updates stream as JSON patches, followed by one terminal task result
• Port/adapter design: Pluggable LLM backends via the CompletionModel protocol
• Durability-ready: TaskState is the complete checkpoint — serialize, store, resume

Installation

uv sync

For development dependencies:

uv sync --all-extras

Quick Start

MISTRAL_API_KEY=<your-key> uv run python examples/demo_simple.py

Simple Agent with Tools

from mistralai.vibe.protocol import AgentTask, ToolTask, TaskState, PendingOutput, MessageEntry, MessageEntryPayload
from mistralai.vibe.protocol.patterns.completion.adapters.mistral import MistralAdapter

def add_numbers(a: float, b: float) -> str:
    return str(a + b)

completion = MistralAdapter(api_key=api_key)
add_tool = ToolTask(fn=add_numbers, name="add_numbers", description="Add two numbers")

agent = AgentTask(
    completion=completion,
    tasks={"add_numbers": add_tool},
    system_prompt="You are a helpful assistant.",
)

state = TaskState(
    id="demo",
    output=PendingOutput(),
    history=[MessageEntry(payload=MessageEntryPayload(role="user", content="What is 3 + 4?"))],
)

channel = await agent.run(state)
async for message in channel:
    # history_update events stream while the task runs,
    # then one terminal task_result carries the final TaskState.
    print(message)

Nested Agents (Subtask Delegation)

researcher = AgentTask(
    completion=completion,
    tasks={"search_web": search_tool},
    name="researcher",
    description="Research a topic using web search",
    system_prompt="Research the given topic, then summarize findings.",
    input_schema={"type": "object", "properties": {"topic": {"type": "string"}}},
)

orchestrator = AgentTask(
    completion=completion,
    tasks={"researcher": researcher},
    system_prompt="Delegate research to the 'researcher' tool, then synthesize.",
)

channel = await orchestrator.run(state)
async for message in channel:
    print(message)

The orchestrator LLM sees researcher as a tool. When it calls researcher(topic="quantum computing"), the runtime spawns a nested agent loop with its own LLM and tools. The parent receives structured output as a tool result.

Documentation

• ARCHITECTURE.md — System design, module structure, and data flow
• documentation/protocol_2026_04_07.md — Current authoritative Task protocol specification
• documentation/INDEX.md — Index of design docs, RFCs, and protocol specs
• presentation/README.md — Slide-ready markdown for SDK architecture explanations

Development

# Type checking
uv run pyright

# Linting
uv run ruff check .

# Tests
uv run pytest tests/

Project Structure

mistralai/vibe/protocol/
  __init__.py              # Package exports
  logging.py               # Structured logging configuration

  types/                   # Pure types and contracts
    state.py               # TaskState, HistoryEntry, TaskOutput
    events.py              # history_update, task_result, callback_* messages
    channel.py             # Channel protocol
    task.py                # Task protocol (@runtime_checkable)
    patch.py               # RFC 6902 JSON Patch ops + AppendOp extension

  modules/                 # Generic engine components
    execution.py           # ExecutionLoop, StateModule ABC, Downstream protocol
    sub_task.py            # sub_task_reducer, subtask event/effect types
    produce.py             # Immer-like produce() + diff()
    json_patch.py          # apply_patches(), reroute_patches()
    task.py                # ModuleTask base class, TaskConfigBase

  patterns/                # Concrete task implementations
    agent.py               # AgentTask, AgentModule (LLM-driven orchestration)
    tool.py                # ToolTask (function wrapper)
    completion/            # LLM completion domain
      port.py              # CompletionModel protocol
      types.py             # CompletionRequest, CompletionChunk
      messages.py          # Message, ToolCall, FunctionCall
      bridge.py            # stream_to_mutations, build_completion_request
      adapters/
        mistral.py         # Mistral API adapter

tests/
  conftest.py              # Shared test factories and mocks
  test_protocol_types.py   # Protocol type tests
  test_produce.py          # produce() / diff() tests
  test_execution_loop.py   # ExecutionLoop tests
  test_replay_invariants.py # Reducer/loop invariant tests
  test_effect_handlers.py  # Standalone effect handler tests
  test_agent_reducer.py    # AgentModule reducer unit tests
  test_sub_task.py         # sub_task_reducer tests
  test_tool_task.py        # ToolTask tests
  test_stream_bridge.py    # LLM streaming bridge tests
  test_agent_task.py       # AgentTask integration tests

examples/
  render.py                # Rendering utilities for example output
  configs/                 # Demo configurations
    simple.yaml            # Basic agent with tools
    subagents.yaml         # Parent with child researchers
    callbacks.yaml         # Callback resolution patterns
    triple_callback.yaml   # 3-level callback chain
    agent_callback.yaml    # Agent-as-callback implementation