Skip to main content

Tools & Agents

The CRP SDK gives LLM calls access to external tools while keeping the same progressive-disclosure design. You register Python functions, the SDK dispatches them when the model asks for a tool, and the results flow back into CKF as structured facts.

---

Register a tool

Use the @client.tool decorator to expose a plain Python function to the model:

import crp

client = crp.SDKClient()

@client.tool
def search_knowledge(query: str) -> str:
    """Search the ingested knowledge base."""
    hits = client.knowledge.search(query, top_k=5)
    return "\n".join(f"- {h['fact']}" for h in hits)

@client.tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    return str(eval(expression, {"__builtins__": {}}, {}))

CRP derives the JSON Schema for the tool from the function signature and docstring. The tool is automatically offered to providers that support tool calling.

---

Automatic tool loop

When you call complete() or ask() and the provider supports tool calling, the SDK will:

1. Send the user message and available tools to the model.
2. If the model requests a tool, execute the registered function.
3. Extract structured facts from the tool result and add them to CKF.
4. Send the tool result back to the model.
5. Repeat until the model produces a final answer.

answer = client.ask(
    "What is 153 * 42, and which docs mention the scheduler?",
    depth="standard",
)

print(answer.text)
print(answer.sources)   # includes tool result citations
print(answer.quality)   # S/A/B/C/D

Tool outputs are treated as first-class context: they are ranked, scored, and packed into subsequent envelopes exactly like ingested documents.

---

Positioned tool loop (v5)

CRP v5 adds dispatch_positioned() - a positioned agentic loop for small and local models (SPEC-049/050). Instead of injecting the whole tool catalogue into the prompt, CRP classifies the request into operations and positions the model on one operation at a time with only the 1–3 tools that operation needs. Each tool result becomes a typed fact in a Cognitive State Object (CSO), so the working window stays bounded no matter how many tools run - positioning, not injection.

import crp

client = crp.SDKClient(provider=my_provider)

@client.tool
def lookup_port_service(port: str) -> dict:
    """Map a TCP port to its well-known service."""
    return {"port": port, "service": {"443": "HTTPS", "22": "SSH"}.get(port, "unknown")}

result = client.dispatch_positioned(
    "Look up the service on port 443, then summarise what it is used for."
)

print(result.text)                 # assembled answer
print(result.operations)           # e.g. ['retrieve', 'synthesise']
print(result.observation_count)    # typed CSO observations stored
print(result.frame_tokens_total)   # bounded per-operation window size
print(result.headers)              # CRP-Agent-Operation-State / -Plan / -Type …
for e in result.event_stream:      # live visibility == audit
    print(e["state"], e["operation"], e["detail"])

dispatch_positioned() returns a PositionedResult with .text, .cso, .event_stream, .operations, .observation_count, .frame_tokens_total, .continuation_windows, .headers, and .halted.

Multi-turn (state relay)

The easiest way to run a multi-turn agent is client.conversation(), which relays the Cognitive State Object forward automatically - no manual state threading:

convo = client.conversation()
r1 = convo.say("Look up the service on port 443.")
r2 = convo.say("Based on what we found, is it safe for a login page?")
print(convo.turns)   # 2

If you prefer explicit control, pass the previous result's .cso as prior_cso:

turn1 = client.dispatch_positioned("Look up the service on port 443.")
turn2 = client.dispatch_positioned(
    "Based on what we found, is it safe for a login page?",
    prior_cso=turn1.cso,          # ← carry state forward
)

Output continuation (long generations)

For long documents, allow the operation to span multiple windows. CRP re-positions each window with the residual task and stops when the output is structurally complete (or the cap is hit). The working prompt stays bounded across all windows:

result = client.dispatch_positioned(
    "Write a comprehensive multi-section guide to Kubernetes networking.",
    max_continuation_windows=10,
)
print(result.continuation_windows)   # e.g. 10
print(result.headers["CRP-Continuation-Windows"])

Safety checkpoints & oversight

Gate destructive or mutating tools behind a human-in-the-loop checkpoint. When the model selects a gated capability, CRP raises a CLARIFY checkpoint; your handler approves or rejects it, and rejection halts safely (never a raw error):

from crp.tools import SafetyClass

def approve(request) -> "ClarificationResolution":
    # show request.question to a human; return their decision
    from crp.security.clarify import ClarificationResolution, ClarificationAction
    return ClarificationResolution(action=ClarificationAction.ANSWER, answer="approve")

result = client.dispatch_positioned(
    "Delete stale records.",
    oversight_required={SafetyClass.DESTRUCTIVE},
    clarify_handler=approve,
)
print(result.halted)   # True if the checkpoint rejected the action

Preventive-safety halts, policy pre-filters (policy=PolicyContext(...)), and resource-governed profiles (governor=ResourceGovernor()) compose the same way.

---

Tool result extraction

After a tool runs, CRP applies a lightweight extraction pipeline to turn the raw output into typed facts:

Stage	What it does
Regex	Dates, identifiers, numbers
NER	Named entities (GLiNER)
Relations	Subject-verb-object triples
Discourse	RST structure for long outputs
Relational	LLM-assisted fact consolidation

These facts are stored in the warm CKF layer and become candidates for future retrieval.

---

Multi-turn agents

For agentic workflows, combine tools with ask() and session persistence:

with crp.SDKClient() as client:
    client.ingest("./docs/")

    # Turn 1: plan
    plan = client.ask("Plan a 10-section Kubernetes troubleshooting guide")

    # Turn 2: execute with tools
    for section in plan.decisions:
        result = client.ask(
            f"Write section: {section}",
            depth="thorough",
        )
        print(result.text)

    # Session status and audit trail
    print(client.session().status())
    print(client.audit.summary())

The session object keeps window history, fact graph, and audit trail across turns.

---

Agent namespace

client.agent exposes multi-agent safety budgets and oversight primitives without requiring you to construct them manually:

budget = client.agent.budget(role="researcher")
print(budget.remaining)
print(budget.allow_dispatch())

Providers namespace

client.providers lets you inspect and select LLM providers at runtime:

print(client.providers.list_available())
adapter = client.providers.get("openai")
print(adapter.context_window_size)

Safety and budgets

Every tool call inherits the active safety profile:

client.configure(safety_profile="strict")

• Input to tools is scanned for PII and injection.
• Tool outputs are checked against the same DPE pipeline as model outputs.
• Multi-agent budgets can be enforced via headers when using the Gateway.

---

What is implemented today

Feature	Status	Notes
@client.tool registration	✅ Implemented	Schema derived from signature
Automatic tool loop	✅ Implemented	Provider tool-calling only
Tool result extraction to CKF	✅ Implemented	Via extraction pipeline
Multi-turn ask()	✅ Implemented	Session state persists
Manual call_tool()	✅ Implemented	For explicit tool execution
Parallel tool calls	⚠️ Partial	Sequential loop today
Streaming tool events	🚧 Not yet	Roadmap
Agent orchestration primitives	🚧 Not yet	Use ask() + sessions today

---

Design rule

CRP never modifies LLM output. Tool results are extracted into CKF, but the original model text is returned untouched. This preserves Axiom 9 (output integrity) while still giving the system structured memory of what the tool produced.

Context Management AI Safety