Tool-Call Efficiency

Overview

Two different kinds of waste show up in an agentic tool-calling loop:

• Discovery waste — before doing anything useful, an agent often calls tools/list, describes a schema, or samples a few rows just to figure out what's available. This repeats on every session even though the answer rarely changes.
• Repetition waste — a recurring task (a daily report, a routine triage flow) re-runs the exact same multi-step tool-call sequence every time, even when the underlying data and the correct sequence of calls haven't changed.

Smartflow's existing per-call semantic cache already helps with individual repeated tool calls. This guide covers four additions layered on top:

Quick Start

from smartflow import SmartflowClient
from smartflow.agent_tools import smartflow_task, smartflow_tool

# Decorate any tool function — sync or async, plain or already wrapped
# by LangChain / LangGraph / CrewAI's own @tool decorator.
@smartflow_tool(name="search_docs")
async def search_docs(query: str) -> str:
    return await my_search_backend(query)

async with SmartflowClient("http://your-smartflow:7775") as sf:
    async with smartflow_task(sf, task_key="daily-report:2026-08-07") as task:
        if task.cached:
            results = task.replay()  # zero live tool calls made
        else:
            results = [await search_docs("Q2 revenue")]
            # ... more tool calls as needed ...
        # commit happens automatically on clean exit from the `async with` block

Trajectory Cache

The trajectory cache memoizes an entire sequence of tool calls that a repeated task historically resolved to — not just one call. smartflow_task is a context manager around it:

1. On entry, it calls lookup_trajectory(task_key). A hit sets task.cached = True and loads the recorded steps — task.replay() returns every step's result in order, with no underlying tool calls made.
2. On a miss, each @smartflow_tool-decorated call inside the block runs normally and is recorded as the next step.
3. On clean exit, the recording is committed — the next call with the same task_key will hit. If the block raises, the recording is discarded instead, so a broken sequence is never cached or replayed.

Deriving a good task_key is the one thing the caller controls — use task_key_for(*parts) for a stable hash of whatever makes two runs "the same task" (a normalized request, a date bucket, a customer ID):

from smartflow.agent_tools import task_key_for

key = task_key_for("daily-report", customer_id, today.isoformat())

Discovery Cache

Every tool schema Smartflow's MCP gateway indexes for semantic tool search is now also available as a direct discovery-cache read — a client can ask "what tools does this server have, and what are their input schemas?" and get an answer without the gateway issuing a live tools/list call.

hit = await sf.discover_tools("github-tools")
if hit["served_from_cache"]:
    tools = hit["tools"]   # each has name, description, input_schema
else:
    # miss — fall back to a live tools/list; it will populate the index
    pass

get_discovery_cache_stats() returns hit/miss counters and how many servers/tools are indexed.

Tool-Call Reduction Benchmark

A live, gateway-computed answer to "how many fewer tool calls are we making," derived directly from the MCP cache's hit/miss counters — a cache hit is a call that never reached the live server, i.e. one fewer round trip.

bench = await sf.get_tool_call_benchmark()
print(f"{bench['pct_calls_avoided']:.0%} fewer tool calls, "
      f"~{bench['estimated_tokens_saved']:,} tokens saved")
print(bench["top_tools"])       # per-tool breakdown
print(bench["methodology"])    # exactly how each number is derived

Compression Quality Delta

Smartflow's semantic compression pipeline can rewrite text — deduplicating repeated concepts, abbreviating, stripping filler. That's lossy by design, and lossy rewrites of enterprise content deserve evidence, not just a compression-ratio number. Pass quality_check=True to also score the compressed output against the original with the HHEM hallucination-eval sidecar:

res = await sf.compress_text(long_text, quality_check=True)
print(f"{res['compression_ratio']:.1f}x compression, "
      f"faithfulness={res.get('quality_delta')}")

quality_delta is 1.0 − hallucination_score — 1.0 means the compressed text stayed fully faithful to the original; lower values mean the rewrite likely introduced unsupported or altered content. It comes back None whenever the check is disabled or the HHEM service is unreachable — never a fabricated default.

Raw HTTP Endpoints

Every SDK method above is a thin wrapper — call these directly from any language.

Scope & Roadmap

• Trajectory matching is exact on task_key today. Semantic/fuzzy task matching (two differently-worded requests that resolve to the same underlying task) is a natural next step, not yet built.
• Discovery cache depends on a server having been indexed at least once via a live tools/list — the very first call for a new server is always a miss.
• estimated_tokens_saved on the reduction benchmark uses a configurable per-call token estimate (TOOL_CALL_AVG_TOKENS_ESTIMATE, default 300), not a per-call measurement — see the returned methodology string.
• Sync wrappers (SyncSmartflowClient) for the new methods are not yet implemented — use SmartflowClient (async) for now.