stream.drivers

stream.drivers

Agent execution backends for Stream.

The portable path has no Cursor dependency: ClaudeDriver and GptDriver (both subclasses of :class:ApiDriver) call provider APIs directly and run the zettelkasten tool-use loop in-process. CursorSdkDriver is an optional, lazily-imported plugin for running inside Cursor.

Use :func:make_driver to construct one by name; the heavy SDK imports happen only when a driver is actually instantiated.

ApiDriver

Provider-agnostic tool-use loop.

Subclasses set name/default_model and implement _format_tools, _complete, _record_assistant, _record_tool_results, and _init_messages.

Source code in stream/drivers/base.py

class ApiDriver:
    """Provider-agnostic tool-use loop.

    Subclasses set ``name``/``default_model`` and implement ``_format_tools``,
    ``_complete``, ``_record_assistant``, ``_record_tool_results``, and
    ``_init_messages``.
    """

    name: str = "api"
    supports_logprobs: bool = False
    default_model: str = ""
    max_tool_iterations: int = 24

    # -- public protocol ----------------------------------------------------

    def run_agent(
        self,
        *,
        role: str,
        persona: str,
        prompt: str,
        tools: list[Tool],
        model: str = "",
        ctx: dict[str, Any] | None = None,
    ) -> AgentResult:
        tool_map = {t.name: t for t in tools}
        tool_specs = self._format_tools(tools)
        chosen_model = model or self.default_model
        messages = self._init_messages(prompt)
        calls = 0
        final_text = ""

        for _ in range(self.max_tool_iterations):
            turn = self._complete(persona, messages, tool_specs, chosen_model)
            self._record_assistant(messages, turn)
            if not turn.tool_calls:
                final_text = turn.text
                break
            results: list[tuple[ToolCall, Any]] = []
            for tc in turn.tool_calls:
                calls += 1
                tool = tool_map.get(tc.name)
                if tool is None:
                    out: Any = {"error": f"unknown tool: {tc.name}"}
                else:
                    try:
                        out = tool.execute(**(tc.input or {}))
                    except Exception as exc:  # surfaced back to the model
                        out = {"error": f"{type(exc).__name__}: {exc}"}
                results.append((tc, out))
            self._record_tool_results(messages, results)
        else:
            # Loop budget exhausted without a tool-free turn.
            final_text = final_text or (
                "RESULT: FAIL\nNOTES: tool-use loop exhausted without completion"
            )

        return AgentResult(text=final_text, ok=True, tool_calls=calls)

    def score(
        self,
        prompt: str,
        *,
        choices: list[str] | None = None,
        model: str = "",
    ) -> Score:
        raise NotImplementedError(
            f"{self.name} driver does not support score(); "
            "use a logprob-capable driver (e.g. GptDriver against vLLM)."
        )

    # -- provider hooks (override) -----------------------------------------

    def _init_messages(self, prompt: str) -> list[Any]:
        raise NotImplementedError

    def _format_tools(self, tools: list[Tool]) -> Any:
        raise NotImplementedError

    def _complete(
        self, system: str, messages: list[Any], tool_specs: Any, model: str
    ) -> ProviderTurn:
        raise NotImplementedError

    def _record_assistant(self, messages: list[Any], turn: ProviderTurn) -> None:
        raise NotImplementedError

    def _record_tool_results(
        self, messages: list[Any], results: list[tuple[ToolCall, Any]]
    ) -> None:
        raise NotImplementedError

ProviderTurn dataclass

One assistant turn parsed into a provider-neutral shape.

Source code in stream/drivers/base.py

@dataclass
class ProviderTurn:
    """One assistant turn parsed into a provider-neutral shape."""

    text: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    raw: Any = None

ToolCall dataclass

A single tool invocation requested by the model.

Source code in stream/drivers/base.py

@dataclass
class ToolCall:
    """A single tool invocation requested by the model."""

    id: str
    name: str
    input: dict[str, Any] = field(default_factory=dict)

make_driver

make_driver(name: str, **kwargs: Any)

Construct a driver by short name.

Recognized: claude, gpt/openai, cursor. Provider SDKs are imported lazily inside each driver module.

Source code in stream/drivers/__init__.py

def make_driver(name: str, **kwargs: Any):
    """Construct a driver by short name.

    Recognized: ``claude``, ``gpt``/``openai``, ``cursor``. Provider SDKs are
    imported lazily inside each driver module.
    """
    key = (name or "claude").lower()
    if key in ("claude", "anthropic"):
        from .claude import ClaudeDriver

        return ClaudeDriver(**kwargs)
    if key in ("gpt", "openai", "vllm", "local"):
        from .gpt import GptDriver

        return GptDriver(**kwargs)
    if key in ("cursor", "cursor-sdk", "cursor_sdk"):
        from .cursor import CursorSdkDriver

        return CursorSdkDriver(**kwargs)
    raise ValueError(
        f"Unknown driver {name!r}. Use one of: claude, gpt, cursor."
    )