Source code for klea_utils.tools
#!/usr/bin/env python3
"""
Tool-related utilities for Klea
File: klea_utils/tools.py
Copyright 2026 Ankur Sinha
Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>
"""
import logging
from typing import Any
from fastmcp.client.client import CallToolResult
from mcp.types import EmbeddedResource, TextContent, Tool
logger = logging.getLogger(__name__)
def _textualize_content_block(block: Any) -> str:
"""Extract text from a single ContentBlock."""
if isinstance(block, TextContent):
return block.text
elif isinstance(block, EmbeddedResource):
resource = block.resource
if hasattr(resource, "blob") and resource.blob:
logger.warning("Blob resource not processed: %s", resource.uri)
text = getattr(resource, "text", None) or getattr(resource, "blob", "")
return f"[Resource: {resource.uri}]\n{text}"
else:
logger.warning("Unhandled content block type: %s", type(block).__name__)
return str(block)
[docs]
def textualize_tool_results(
tool_results: list[CallToolResult],
) -> str:
"""Format tool call results as LLM-ready text for use in prompt context.
Tool return values are typically JSON-structured data (dicts, lists) which
LLMs handle naturally. We wrap them in markdown code blocks for clear
separation from surrounding prompt text.
:param tool_results: List of tool call results
:returns: Formatted string suitable for inclusion in an LLM prompt
"""
if not tool_results:
return ""
text = "## Tool Results\n"
for i, result in enumerate(tool_results, 1):
text += f"\n### Result {i}/{len(tool_results)}\n"
if result.is_error:
parts = [_textualize_content_block(c) for c in result.content]
text += "**Error:** " + "\n".join(parts) + "\n"
else:
parts = [_textualize_content_block(c) for c in result.content]
text += "```\n" + "\n".join(parts) + "\n```\n"
return text
def _collapse_schema_type(schema: dict[str, Any]) -> str:
"""Collapse a JSON Schema type spec into a single compact type name.
Handles ``anyOf`` unions (e.g. ``[{"type": "integer"}, {"type": "null"}]``
collapses to ``integer``) and plain ``type`` strings.
"""
if "anyOf" in schema:
types = [
t.get("type")
for t in schema["anyOf"]
if isinstance(t, dict) and t.get("type") != "null"
]
if types:
return "|".join(str(t) for t in types)
return schema.get("type") or "any"
def _format_tool_parameters(input_schema: dict[str, Any] | None) -> str:
"""Return a compact one-line-per-parameter summary of an MCP tool schema.
The raw JSON Schema ``properties`` dict is verbose (defaults, length and
range validators, ``anyOf`` unions); only the parameter name, type,
required flag, and description are useful for tool selection, so the
rest is dropped and whitespace is normalised.
"""
if not input_schema:
return ""
properties = input_schema.get("properties") or {}
if not properties:
return ""
required = set(input_schema.get("required") or [])
lines = []
for name, schema in properties.items():
if not isinstance(schema, dict):
continue
ptype = _collapse_schema_type(schema)
desc = " ".join((schema.get("description") or "").split())
flag = ", required" if name in required else ""
lines.append(f"- {name} ({ptype}{flag}): {desc}".rstrip())
if not lines:
return ""
return "Parameters:\n" + "\n".join(lines)