Example — MCP & tool results

When your context comes from calling a tool (an MCP tool or a REST API) rather than a document store, the "source" is a live result. Coalent caches the understanding built from it and keeps it fresh — you just derive a stable artifact_id from the call.

1. A tool-backed retriever

This isn't a vector lookup, so we implement the Retriever interface directly — just a class with a retrieve method (no base class; Retriever is a structural protocol):

Derive artifact_id from the tool name + its arguments, so the same call always maps to the same cached source — never a hardcoded string:

import hashlib, json
from coalent import Chunk

def tool_artifact_id(tool: str, args: dict) -> str:
    key = hashlib.sha1(json.dumps(args, sort_keys=True).encode()).hexdigest()[:12]
    return f"tool:{tool}:{key}"            # e.g. tool:get_inventory:9f1c2a...

class ToolRetriever:
    def __init__(self, call_tool):
        self._call = call_tool             # your MCP/API client

    def retrieve(self, query, *, namespace=None):
        tool, args = plan_call(query)      # however you decide which tool to call
        result = self._call(tool, args)    # the live tool/API result
        return [Chunk(
            artifact_id=tool_artifact_id(tool, args),   # dynamic — from the call
            text=result["text"],
            version=str(result.get("revision", "")),
        )]

Already structured? Skip the LLM.

If the tool returns JSON that's already decision-ready, pair this retriever with JSONPassthroughSynthesizer — it caches the JSON as the understanding (no model call, no latency), while still capturing provenance so the invalidation below works unchanged:

from coalent import SemanticCache, JSONPassthroughSynthesizer

cache = SemanticCache(ToolRetriever(call_tool), JSONPassthroughSynthesizer())

Reach for an LLMSynthesizer instead only when the tool returns prose that needs to be understood.

2. Invalidate — two honest options

Tool results are volatile and usually have no change feed, so pick the pattern that fits:

Option A — invalidate at the write site (you own the mutation)

When your app changes the underlying entity, bust it right there with the same id derivation:

def update_inventory(sku, qty):
    db.update(sku, qty)                                       # the write
    cache.source_changed(tool_artifact_id("get_inventory", {"sku": sku}))

Option B — TTL + revalidate (changes happen elsewhere)

No feed? Revalidate on a TTL — re-call the tool and hash the result. Unchanged → stays fresh (no rebuild); changed → re-materializes:

from coalent import SemanticCache, FreshnessPolicy

def revalidate(artifact_id: str) -> tuple[str, str]:
    tool, args = parse_artifact_id(artifact_id)   # recover the call from the id
    result = call_tool(tool, args)
    return result["text"], str(result.get("revision", ""))

cache = SemanticCache(
    ToolRetriever(call_tool),
    synthesizer,
    freshness=FreshnessPolicy(max_age=60, revalidate=revalidate),  # seconds
)

Encode enough in the artifact_id (or a side table) to recover the call for revalidation. A readable scheme like tool:get_inventory:<args-hash> plus a small {hash: args} map works well.

The boundary

Coalent caches and freshens context — it does not call your tools for actions. Tool calls for retrieval live in your retrieve() / revalidate; tool calls that do things stay in your agent. Clean separation.