stream.tools

stream.tools

In-process zettelkasten tool registry shared by all API drivers.

These are the function tools an agent can call during its turn. They wrap the importable zettelkasten.server functions directly (no MCP round-trip), so a portable driver talks to the store in-process. The zettelkasten import is lazy: import stream.tools stays cheap and does not require the store to be present.

Tools are tagged write=True when they mutate the graph. The executor strips write tools for read-only roles (planner/checker/meta) via :func:tools_for_role so only implementers (scribe/engineer) can change anything.

zettelkasten_tools

zettelkasten_tools() -> list[Tool]

Return the full curated set of zettelkasten function tools.

Source code in stream/tools.py

def zettelkasten_tools() -> list[Tool]:
    """Return the full curated set of zettelkasten function tools."""
    return [
        Tool(
            name="search_notes",
            description=(
                "Semantic + lexical search for existing notes in a concept box. "
                "Use before adding a note to dedup against what already exists."
            ),
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string", "description": "Concept box / graph name"},
                    "query": {"type": "string"},
                    "top_k": {"type": "integer", "default": 8},
                },
                "required": ["graph", "query"],
            },
            execute=_call("search_notes"),
        ),
        Tool(
            name="find_by_title",
            description="Find a note by its exact title within a graph.",
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string"},
                    "title": {"type": "string"},
                },
                "required": ["graph", "title"],
            },
            execute=_call("find_by_title"),
        ),
        Tool(
            name="get_note",
            description="Fetch a single note by id (or title) with its full body and metadata.",
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string"},
                    "note_id": {"type": "string"},
                    "title": {"type": "string"},
                },
                "required": ["graph"],
            },
            execute=_call("get_note"),
        ),
        Tool(
            name="get_fulltext",
            description=(
                "Return the extracted full text of a source (capped at max_chars, "
                "0 = no cap). Copy quotes verbatim from here so they ground."
            ),
            input_schema={
                "type": "object",
                "properties": {
                    "source": {"type": "string", "description": "Source graph name or id"},
                    "max_chars": {"type": "integer", "default": 0},
                },
                "required": ["source"],
            },
            execute=_fulltext_execute,
        ),
        Tool(
            name="suggest_connections",
            description="Suggest existing notes that a given note should link to.",
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string"},
                    "note_id": {"type": "string"},
                    "top_k": {"type": "integer", "default": 8},
                },
                "required": ["graph", "note_id"],
            },
            execute=_call("suggest_connections"),
        ),
        Tool(
            name="add_note",
            description=(
                "Add a new note to a concept box. Always search_notes first to "
                "avoid duplicates. Quotes must be verbatim from get_fulltext."
            ),
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string"},
                    "title": {"type": "string"},
                    "type": {
                        "type": "string",
                        "description": (
                            "claim, concept, critique, definition, example, finding, "
                            "method, model, question, quote, or synthesis"
                        ),
                    },
                    "body": {"type": "string"},
                    "source_book": {"type": "string"},
                    "source_page": {"type": "string"},
                    "tags": {"type": "array", "items": {"type": "string"}},
                    "links": {
                        "type": "array",
                        "items": {"type": "object"},
                        "description": "List of {target, relation, direction?}",
                    },
                    "prerequisites": {"type": "array", "items": {"type": "string"}},
                    "aliases": {"type": "array", "items": {"type": "string"}},
                    "project": {"type": "string"},
                    "synthesis_status": {"type": "string"},
                },
                "required": ["graph", "title", "type", "body"],
            },
            execute=_call("add_note"),
            write=True,
        ),
        Tool(
            name="update_note",
            description="Update fields of an existing note (body, tags, links, ...).",
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string"},
                    "note_id": {"type": "string"},
                    "title": {"type": "string"},
                    "body": {"type": "string"},
                    "tags": {"type": "array", "items": {"type": "string"}},
                    "add_links": {"type": "array", "items": {"type": "object"}},
                    "add_prerequisites": {"type": "array", "items": {"type": "string"}},
                    "add_aliases": {"type": "array", "items": {"type": "string"}},
                },
                "required": ["graph", "note_id"],
            },
            execute=_call("update_note"),
            write=True,
        ),
        Tool(
            name="link_notes",
            description="Create a typed link between two existing notes.",
            input_schema={
                "type": "object",
                "properties": {
                    "graph": {"type": "string"},
                    "source_id": {"type": "string"},
                    "target_id": {"type": "string"},
                    "relation": {"type": "string"},
                    "direction": {"type": "string"},
                },
                "required": ["graph", "source_id", "target_id", "relation"],
            },
            execute=_call("link_notes"),
            write=True,
        ),
        Tool(
            name="add_to_project",
            description="Register a source graph into a project's membership list.",
            input_schema={
                "type": "object",
                "properties": {
                    "project": {"type": "string"},
                    "source": {"type": "string"},
                },
                "required": ["project", "source"],
            },
            execute=_call("add_to_project"),
            write=True,
        ),
    ]

tools_for_role

tools_for_role(role: str, tools: list[Tool] | None = None) -> list[Tool]

Filter the registry for a workflow role.

Implementers get the full set; read-only roles (planner/checker/meta) get the read subset only.

Source code in stream/tools.py

def tools_for_role(role: str, tools: list[Tool] | None = None) -> list[Tool]:
    """Filter the registry for a workflow role.

    Implementers get the full set; read-only roles (planner/checker/meta) get
    the read subset only.
    """
    registry = tools if tools is not None else zettelkasten_tools()
    if role in _WRITE_ROLES:
        return list(registry)
    return [t for t in registry if not t.write]