Skip to content

ontocast.integrations.langchain

Expose OntoCast capabilities as LangChain tools.

ontocast_tools(tools) returns a list of BaseTool objects that any LangChain or LangGraph agent can call:

from langchain.agents import create_agent
from ontocast import Config, ToolBox, ontocast_tools

tools = await ToolBox.acreate(Config.in_memory())
await tools.initialize()

agent = create_agent(model, tools=[*ontocast_tools(tools)])

Two design rules run through this module.

Capability gating. A tool whose backend is missing is not returned, rather than returned and made to fail on first call. A base install has no Qdrant, no docling, and possibly no SPARQL-capable store, and an agent handed a tool that always errors will keep retrying it. :func:ontocast_tool_diagnostics explains each omission, since "my agent has six tools instead of eleven" is otherwise unbreakable.

Mutation is opt-in. The write tools are excluded unless mutating=True. ontocast_delete_ontology drops a named graph, unlinks a file from disk, and deletes vectors -- three irreversible effects from one model-chosen string.

ontocast_tool_diagnostics(tools)

Explain why each unavailable tool is unavailable.

Parameters:

Name Type Description Default
tools 'ToolBox'

The ToolBox the tools would be built against.

required

Returns:

Type Description
dict[str, str]

Mapping of tool name to the reason it would be skipped. Tools that are

dict[str, str]

available are absent from the mapping.

Source code in ontocast/integrations/langchain.py
def ontocast_tool_diagnostics(tools: "ToolBox") -> dict[str, str]:
    """Explain why each unavailable tool is unavailable.

    Args:
        tools: The ToolBox the tools would be built against.

    Returns:
        Mapping of tool name to the reason it would be skipped. Tools that are
        available are absent from the mapping.
    """
    reasons: dict[str, str] = {}
    for name in ALL_TOOL_NAMES:
        reason = _unavailable_reason(name, tools)
        if reason is not None:
            reasons[name] = reason
    return reasons

ontocast_tool_names(tools, *, mutating=False)

Return the names :func:ontocast_tools would produce, without building them.

Source code in ontocast/integrations/langchain.py
def ontocast_tool_names(
    tools: "ToolBox",
    *,
    mutating: bool = False,
) -> list[str]:
    """Return the names :func:`ontocast_tools` would produce, without building them."""
    requested = _resolve_requested(None, None, mutating)
    return [
        name
        for name in ALL_TOOL_NAMES
        if name in requested and _unavailable_reason(name, tools) is None
    ]

ontocast_tools(tools, *, include=None, exclude=None, mutating=False, max_chars=20000)

Wrap OntoCast capabilities as LangChain structured tools.

Only tools whose backend is installed and configured are returned; call :func:ontocast_tool_diagnostics to see why something is missing.

All tools are async-only. Agents must invoke them with ainvoke; a synchronous invoke raises NotImplementedError. Several of the underlying calls are coroutines already, and the rest are CPU-heavy enough that running them on the caller's event loop would stall it.

Parameters:

Name Type Description Default
tools 'ToolBox'

A constructed ToolBox. Call await tools.initialize() first if you want the ontology catalog populated.

required
include Iterable[str] | None

Restrict to these tool names. None selects the default set (read tools, plus mutating tools when mutating is true). Naming a tool here also opts into the OPT_IN_TOOLS.

None
exclude Iterable[str] | None

Drop these names from whatever include selected.

None
mutating bool

Include the write tools. Off by default; each one changes stored state irreversibly.

False
max_chars int

Truncation budget applied to each tool's rendered result.

20000

Returns:

Type Description
list[BaseTool]

Available tools in a stable order.

Raises:

Type Description
ValueError

If include or exclude names an unknown tool.

Source code in ontocast/integrations/langchain.py
def ontocast_tools(
    tools: "ToolBox",
    *,
    include: Iterable[str] | None = None,
    exclude: Iterable[str] | None = None,
    mutating: bool = False,
    max_chars: int = 20_000,
) -> list[BaseTool]:
    """Wrap OntoCast capabilities as LangChain structured tools.

    Only tools whose backend is installed and configured are returned; call
    :func:`ontocast_tool_diagnostics` to see why something is missing.

    All tools are async-only. Agents must invoke them with ``ainvoke``; a
    synchronous ``invoke`` raises ``NotImplementedError``. Several of the
    underlying calls are coroutines already, and the rest are CPU-heavy enough
    that running them on the caller's event loop would stall it.

    Args:
        tools: A constructed ToolBox. Call ``await tools.initialize()`` first if
            you want the ontology catalog populated.
        include: Restrict to these tool names. ``None`` selects the default set
            (read tools, plus mutating tools when ``mutating`` is true).
            Naming a tool here also opts into the ``OPT_IN_TOOLS``.
        exclude: Drop these names from whatever ``include`` selected.
        mutating: Include the write tools. Off by default; each one changes
            stored state irreversibly.
        max_chars: Truncation budget applied to each tool's rendered result.

    Returns:
        Available tools in a stable order.

    Raises:
        ValueError: If ``include`` or ``exclude`` names an unknown tool.
    """
    requested = _resolve_requested(include, exclude, mutating)
    builders = _builders(tools, max_chars=max_chars)

    built: list[BaseTool] = []
    for name in ALL_TOOL_NAMES:
        if name not in requested:
            continue
        reason = _unavailable_reason(name, tools)
        if reason is not None:
            logger.info("Skipping %s: %s", name, reason)
            continue
        built.append(builders[name]())
    return built