Tool System

ContextRouter exposes tools for LLM function calling. Tools allow agents to interact with Brain (search), Commerce (products), databases, and external services.

Built-in Tools

Tool	Module	Description
Brain Search	`modules/tools/`	Query Brain knowledge store
Commerce Products	`modules/tools/`	Search/update product catalog
Web Connector	`modules/connectors/web.py`	Web search via Google CSE
RSS Connector	`modules/connectors/rss.py`	RSS feed ingestion
File Connector	`modules/connectors/file.py`	Local file reading
API Connector	`modules/connectors/api.py`	Generic HTTP API calls

ToolExecutorStream (Bi-Directional Binding)

In typical LLM applications, the orchestrator (Router) needs direct access to databases or APIs to execute tools. This creates a massive security liability, as the central Mind would need the passwords for every connected project (the Confused Deputy problem).

To solve this, ContextRouter introduced the Bi-Directional Tool Binding via the ToolExecutorStream. External projects connect to the Router which acts purely as a “Cognition Provider”. When the LLM decides to run a tool like sql_query, the Router does not execute it. Instead, it streams the tool request back down the persistent connection to the client project. The client executes the query securely inside its own VPC, and streams the result back up to the LLM.

See also: ContextUnity Security Scope

router-tools

Client-Side Implementation

from contextunity.core import ContextUnit, create_channel_sync, contextunit_pb2
from contextunity.core import router_pb2_grpc

async def tool_executor():
    channel = create_channel_sync("router:50052")
    stub = router_pb2_grpc.RouterServiceStub(channel)

    async for message_pb in stub.ToolExecutorStream(request_stream()):
        message = ContextUnit.from_protobuf(message_pb)
        # Execute tool locally
        tool_name = message.payload.get("tool_name")
        if tool_name == "sql_query":
            result = await run_sql(message.payload["arguments"]["query"])
            response = ContextUnit(
                payload={"result": result, "tool_call_id": message.payload["tool_call_id"]},
            )
            yield response.to_protobuf(contextunit_pb2)

Tool Registration

Tools are registered via the ToolFactory:

from contextunity.router.service.tool_factory import ToolFactory

factory = ToolFactory(config)
tools = factory.build_tools(agent_config)

The factory reads agent configuration to determine which tools are available for each agent, enforcing the principle of least privilege.