Function _web_fetch_call

src/tool_system/tools/web_fetch.py:447–491

(tool_input: dict[str, Any], context: ToolContext)

def _web_fetch_call(tool_input: dict[str, Any], context: ToolContext) -> ToolResult:
    url = tool_input.get("url", "")
    prompt = tool_input.get("prompt", "")
    fmt = tool_input.get("format") or "markdown"
    if fmt not in _VALID_FORMATS:
        fmt = "markdown"

    if not isinstance(url, str) or not url:
        raise ToolInputError("url must be a non-empty string")

    url = _validate_url(url)

    start_time = time.time()

    # Cache the *converted* content; key by format so a markdown fetch and a text
    # fetch of the same URL don't collide.
    cache_key = f"{fmt}:{url}"
    cached = _cache_get(cache_key)
    if cached:
        content, content_type, status = cached
    else:
        raw, content_type, status = _fetch_with_redirect_handling(url, fmt=fmt)
        content = _convert(raw, content_type, fmt)
        _cache_set(cache_key, content, content_type, status)

    if len(content) > 100_000:
        content = content[:100_000] + "\n\n... [truncated] ..."

    duration_ms = int((time.time() - start_time) * 1000)

    result_text = content
    if prompt and isinstance(prompt, str):
        result_text = f"User prompt: {prompt}\n\nContent from {url}:\n\n{content}"

    return ToolResult(
        name="WebFetch",
        output={
            "url": url,
            "content_type": content_type,
            "result": result_text,
            "bytes": len(content.encode("utf-8")),
            "code": status,
            "duration_ms": duration_ms,
        },
    )


# -- Prompt --------------------------------------------------------------------