Documentation Index
Fetch the complete documentation index at: https://mintlify.com/5unnykum4r/grip-ai/llms.txt
Use this file to discover all available pages before exploring further.
Overview
Grip AI’s tool system provides a unified abstraction for both built-in tools and MCP (Model Context Protocol) tools. Every tool implements the Tool ABC and is registered in a central ToolRegistry.
The registry handles:
- Registration — Add/remove tools at runtime
- Schema generation — Export OpenAI function-calling definitions
- Execution dispatch — Route tool calls to the correct implementation
- Result serialization — Convert Pydantic models to JSON automatically
All tools implement this abstract base class:
class Tool(ABC):
@property
@abstractmethod
def name(self) -> str:
"""Unique identifier used in tool_call function_name."""
...
@property
@abstractmethod
def description(self) -> str:
"""One-line description shown to the LLM."""
...
@property
@abstractmethod
def parameters(self) -> dict[str, Any]:
"""JSON Schema (type: object) describing accepted parameters."""
...
@property
def category(self) -> str:
"""Tool category for grouped display. Defaults to 'general'."""
return "general"
@abstractmethod
async def execute(self, params: dict[str, Any], ctx: ToolContext) -> ToolResult:
"""Run the tool with validated parameters and return a result.
Can return a plain string or a Pydantic BaseModel instance.
BaseModel instances are automatically serialized to JSON.
"""
...
def to_definition(self) -> dict[str, Any]:
"""Serialize this tool to OpenAI function-calling schema format."""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
Tool Context
Every tool execution receives a ToolContext with runtime information:
@dataclass
class ToolContext:
workspace_path: Path
restrict_to_workspace: bool = False
shell_timeout: int = 60
session_key: str = ""
extra: dict[str, Any] = field(default_factory=dict)
The extra dict can contain:
brave_api_key — For web search tools
dry_run — Skip actual execution (testing mode)
trust_manager — For sandbox file access validation
The ToolRegistry manages all registered tools:
class ToolRegistry:
def __init__(self) -> None:
self._tools: dict[str, Tool] = {}
self._category_cache: dict[str, list[Tool]] | None = None
self.mcp_manager: Any = None
def register(self, tool: Tool) -> None:
if tool.name in self._tools:
logger.warning("Overwriting existing tool registration: {}", tool.name)
self._tools[tool.name] = tool
self._category_cache = None
logger.debug("Registered tool: {}", tool.name)
def register_many(self, tools: list[Tool]) -> None:
for tool in tools:
self.register(tool)
def get_definitions(self) -> list[dict[str, Any]]:
"""Return OpenAI function-calling definitions for all registered tools."""
return [tool.to_definition() for tool in self._tools.values()]
async def execute(self, name: str, params: dict[str, Any], ctx: ToolContext) -> str:
"""Look up a tool by name and execute it.
Returns error string (not exception) if tool is not found or fails.
Pydantic BaseModel results are serialized to indented JSON automatically.
"""
tool = self._tools.get(name)
if tool is None:
return f"Error: Unknown tool '{name}'. Available: {', '.join(self._tools.keys())}"
try:
result = await tool.execute(params, ctx)
return _serialize_result(result)
except Exception as exc:
logger.error("Unhandled error in tool {}: {}", name, exc, exc_info=True)
return f"Error executing {name}: {type(exc).__name__}: {exc}"
Grip includes tools across multiple categories:
- read_file — Read file contents with offset/limit
- write_file — Write or overwrite file
- append_file — Append to existing file
- list_directory — List files and directories
- create_directory — Create directory tree
- delete_file — Remove file
- move_file — Move or rename file
- search_files — Glob pattern search
- grep_files — Content search with regex
- execute_command — Run shell commands with timeout
- get_environment — Read environment variables
- fetch_url — HTTP GET with headers
- brave_search — Web search via Brave API
- scrape_page — Extract clean text from HTML
- send_message — Send text to user via channel
- send_file — Send file attachment
- spawn_subagent — Launch parallel agent tasks
- schedule_task — Schedule cron jobs
- workflow_execute — Run multi-step workflows
- stock_quote — Get stock price (requires yfinance)
- crypto_price — Get cryptocurrency price
- research_topic — Multi-source research with citations
- fact_check — Verify claims against sources
- analyze_code — Static analysis and metrics
- find_definition — Locate class/function definitions
- trace_calls — Build call graphs
- convert_format — CSV ↔ JSON ↔ YAML conversions
- filter_data — Query JSON with JSONPath
- aggregate_data — Sum, average, group-by operations
- generate_markdown — Create formatted markdown
- generate_pdf — Convert markdown to PDF
- generate_diagram — Create Mermaid diagrams
- compose_email — Draft email with template
- send_email — Send via SMTP (requires config)
- todo_write — Create and update tasks
- todo_read — List active tasks
When the LLM decides to use tools, it returns:
{
"content": "I'll search the codebase for the config file.",
"tool_calls": [
{
"id": "call_abc123",
"function_name": "search_files",
"arguments": {
"pattern": "**/config*.py",
"directory": "/workspace"
}
}
]
}
The agent loop executes all tool calls in parallel:
# Execute all tool calls in parallel via asyncio.gather
exec_results = await asyncio.gather(
*(self._execute_tool(tc, tool_ctx) for tc in response.tool_calls)
)
async def _execute_tool(self, tool_call: ToolCall, ctx: ToolContext) -> ToolExecutionResult:
args = tool_call.arguments if isinstance(tool_call.arguments, dict) else {}
logger.info(
"Executing tool: {}({})",
tool_call.function_name,
", ".join(f"{k}={v!r}" for k, v in list(args.items())[:3]),
)
start = time.perf_counter()
# Prefer ToolRegistry
if self._registry:
output = await self._registry.execute(tool_call.function_name, args, ctx)
elapsed = (time.perf_counter() - start) * 1000
success = not output.startswith("Error:")
return ToolExecutionResult(
tool_call_id=tool_call.id,
tool_name=tool_call.function_name,
output=output,
success=success,
duration_ms=elapsed,
)
4. Result Scrubbing
Secrets are redacted before appending to message history:
for exec_result in exec_results:
all_tool_calls.append(exec_result.tool_name)
all_tool_details.append(
ToolCallDetail(
name=exec_result.tool_name,
success=exec_result.success,
duration_ms=exec_result.duration_ms,
output_preview=exec_result.output[:120],
)
)
# Scrub secrets before storing tool output in message history
scrubbed_output = _scrub_secrets(exec_result.output)
messages.append(
LLMMessage(
role="tool",
content=scrubbed_output,
tool_call_id=exec_result.tool_call_id,
name=exec_result.tool_name,
)
)
5. Loop Continues
The agent loop sends tool results back to the LLM, which can:
- Return final text response (loop ends)
- Make more tool calls (loop continues)
- Hit max_tool_iterations limit (forced completion)
Example custom tool:
from grip.tools.base import Tool, ToolContext, ToolResult
from typing import Any
class MyCustomTool(Tool):
@property
def name(self) -> str:
return "my_custom_tool"
@property
def description(self) -> str:
return "Does something custom and useful"
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"input_text": {
"type": "string",
"description": "The text to process",
},
"format": {
"type": "string",
"enum": ["json", "yaml", "xml"],
"description": "Output format",
},
},
"required": ["input_text"],
}
@property
def category(self) -> str:
return "custom"
async def execute(self, params: dict[str, Any], ctx: ToolContext) -> ToolResult:
input_text = params["input_text"]
format_type = params.get("format", "json")
# Your custom logic here
result = f"Processed {len(input_text)} chars in {format_type} format"
return result
# Register the tool
registry = create_default_registry()
registry.register(MyCustomTool())
MCP servers provide additional tools via the MCP protocol:
class MCPManager:
async def connect_all(
self, mcp_servers: dict[str, MCPServerConfig], registry: ToolRegistry
) -> None:
for name, config in mcp_servers.items():
if not config.enabled:
continue
try:
# Connect to MCP server (stdio or SSE)
client = await self._connect_server(name, config)
# List available tools
tools_result = await client.list_tools()
# Register each tool as a dynamic tool in the registry
for tool_info in tools_result.tools:
mcp_tool = MCPTool(
server_name=name,
tool_name=tool_info.name,
description=tool_info.description,
parameters=tool_info.inputSchema,
client=client,
)
registry.register(mcp_tool)
logger.info(
"Connected to MCP server '{}': {} tools",
name,
len(tools_result.tools),
)
except Exception as exc:
logger.error("Failed to connect to MCP server '{}': {}", name, exc)
MCP tools are registered with prefixed names: mcp__server__tool
Tools are grouped by category for system prompt generation:
def get_tools_by_category(self) -> dict[str, list[Tool]]:
groups: dict[str, list[Tool]] = {}
for tool in self._tools.values():
groups.setdefault(tool.category, []).append(tool)
return groups
Valid categories:
filesystem — File I/O operations
shell — Command execution
web — HTTP requests and scraping
messaging — User communication
orchestration — Subagents, workflows, scheduling
finance — Stock/crypto data
research — Multi-source research
code_analysis — Static analysis
data_transform — Data format conversions
document_gen — Document creation
general — Uncategorized
Configuration
tools:
# Built-in tool settings
restrict_to_workspace: true # Sandbox file access
shell_timeout: 60 # Command timeout in seconds
# Web tools
web:
brave:
enabled: true
api_key: "your-brave-api-key"
# MCP servers
mcp_servers:
filesystem:
enabled: true
command: "uvx"
args: ["mcp-server-filesystem", "/workspace"]
fetch:
enabled: true
command: "uvx"
args: ["mcp-server-fetch"]
allowed_tools:
- "mcp__fetch__fetch"
memory:
enabled: true
url: "https://memory.example.com/sse"
type: "sse"
headers:
Authorization: "Bearer token"
Tools can return strings or Pydantic models:
def _serialize_result(result: ToolResult) -> str:
if isinstance(result, str):
return result
if PydanticBaseModel is not None and isinstance(result, PydanticBaseModel):
return result.model_dump_json(indent=2)
if isinstance(result, (dict, list)):
return json.dumps(result, indent=2, default=str)
return str(result)
Example Pydantic result:
from pydantic import BaseModel
class SearchResult(BaseModel):
total_files: int
matches: list[str]
time_ms: float
class SearchTool(Tool):
async def execute(self, params: dict[str, Any], ctx: ToolContext) -> ToolResult:
# ... search logic ...
return SearchResult(
total_files=len(files),
matches=matching_files,
time_ms=elapsed,
)
The registry automatically serializes to JSON:
{
"total_files": 142,
"matches": [
"/workspace/config.py",
"/workspace/settings/config.yaml"
],
"time_ms": 234.5
}
Next Steps