feat(python): add zeroclaw-tools companion package for LangGraph tool calling

- Add Python package with LangGraph-based agent for consistent tool calling
- Provides reliable tool execution for providers with inconsistent native support
- Includes tools: shell, file_read, file_write, web_search, http_request, memory
- Discord bot integration included
- CLI tool for quick interactions
- Works with any OpenAI-compatible provider (Z.AI, OpenRouter, Groq, etc.)

Why: Some LLM providers (e.g., GLM-5/Zhipu) have inconsistent tool calling behavior.
LangGraph's structured approach guarantees reliable tool execution across all providers.
This commit is contained in:
ZeroClaw Contributor 2026-02-17 01:35:40 +03:00 committed by Chummy
parent bc38994867
commit e5ef8a3b62
17 changed files with 1371 additions and 0 deletions

View file

@ -0,0 +1,20 @@
"""
Built-in tools for ZeroClaw agents.
"""
from .base import tool
from .shell import shell
from .file import file_read, file_write
from .web import web_search, http_request
from .memory import memory_store, memory_recall
__all__ = [
"tool",
"shell",
"file_read",
"file_write",
"web_search",
"http_request",
"memory_store",
"memory_recall",
]

View file

@ -0,0 +1,46 @@
"""
Base utilities for creating tools.
"""
from typing import Any, Callable, Optional
from langchain_core.tools import tool as langchain_tool
def tool(
func: Optional[Callable] = None,
*,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Any:
"""
Decorator to create a LangChain tool from a function.
This is a convenience wrapper around langchain_core.tools.tool that
provides a simpler interface for ZeroClaw users.
Args:
func: The function to wrap (when used without parentheses)
name: Optional custom name for the tool
description: Optional custom description
Returns:
A BaseTool instance
Example:
```python
from zeroclaw_tools import tool
@tool
def my_tool(query: str) -> str:
\"\"\"Description of what this tool does.\"\"\"
return f"Result: {query}"
```
"""
if func is not None:
return langchain_tool(func)
def decorator(f: Callable) -> Any:
return langchain_tool(f, name=name)
return decorator

View file

@ -0,0 +1,60 @@
"""
File read/write tools.
"""
import os
from langchain_core.tools import tool
MAX_FILE_SIZE = 100_000
@tool
def file_read(path: str) -> str:
"""
Read the contents of a file at the given path.
Args:
path: The file path to read (absolute or relative)
Returns:
The file contents, or an error message
"""
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
content = f.read()
if len(content) > MAX_FILE_SIZE:
return content[:MAX_FILE_SIZE] + f"\n... (truncated, {len(content)} bytes total)"
return content
except FileNotFoundError:
return f"Error: File not found: {path}"
except PermissionError:
return f"Error: Permission denied: {path}"
except Exception as e:
return f"Error: {e}"
@tool
def file_write(path: str, content: str) -> str:
"""
Write content to a file, creating directories if needed.
Args:
path: The file path to write to
content: The content to write
Returns:
Success message or error
"""
try:
parent = os.path.dirname(path)
if parent:
os.makedirs(parent, exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return f"Successfully wrote {len(content)} bytes to {path}"
except PermissionError:
return f"Error: Permission denied: {path}"
except Exception as e:
return f"Error: {e}"

View file

@ -0,0 +1,86 @@
"""
Memory storage tools for persisting data between conversations.
"""
import json
import os
from pathlib import Path
from langchain_core.tools import tool
def _get_memory_path() -> Path:
"""Get the path to the memory storage file."""
return Path.home() / ".zeroclaw" / "memory_store.json"
def _load_memory() -> dict:
"""Load memory from disk."""
path = _get_memory_path()
if not path.exists():
return {}
try:
with open(path, "r") as f:
return json.load(f)
except Exception:
return {}
def _save_memory(data: dict) -> None:
"""Save memory to disk."""
path = _get_memory_path()
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w") as f:
json.dump(data, f, indent=2)
@tool
def memory_store(key: str, value: str) -> str:
"""
Store a key-value pair in persistent memory.
Args:
key: The key to store under
value: The value to store
Returns:
Confirmation message
"""
try:
data = _load_memory()
data[key] = value
_save_memory(data)
return f"Stored: {key}"
except Exception as e:
return f"Error: {e}"
@tool
def memory_recall(query: str) -> str:
"""
Search memory for entries matching the query.
Args:
query: The search query
Returns:
Matching entries or "no matches" message
"""
try:
data = _load_memory()
if not data:
return "No memories stored yet"
query_lower = query.lower()
matches = {
k: v
for k, v in data.items()
if query_lower in k.lower() or query_lower in str(v).lower()
}
if not matches:
return f"No matches for: {query}"
return json.dumps(matches, indent=2)
except Exception as e:
return f"Error: {e}"

View file

@ -0,0 +1,32 @@
"""
Shell execution tool.
"""
import subprocess
from langchain_core.tools import tool
@tool
def shell(command: str) -> str:
"""
Execute a shell command and return the output.
Args:
command: The shell command to execute
Returns:
The command output (stdout and stderr combined)
"""
try:
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=60)
output = result.stdout
if result.stderr:
output += f"\nSTDERR: {result.stderr}"
if result.returncode != 0:
output += f"\nExit code: {result.returncode}"
return output or "(no output)"
except subprocess.TimeoutExpired:
return "Error: Command timed out after 60 seconds"
except Exception as e:
return f"Error: {e}"

View file

@ -0,0 +1,88 @@
"""
Web-related tools: HTTP requests and web search.
"""
import json
import os
import urllib.error
import urllib.parse
import urllib.request
from langchain_core.tools import tool
@tool
def http_request(url: str, method: str = "GET", headers: str = "", body: str = "") -> str:
"""
Make an HTTP request to a URL.
Args:
url: The URL to request
method: HTTP method (GET, POST, PUT, DELETE, etc.)
headers: Comma-separated headers in format "Name: Value, Name2: Value2"
body: Request body for POST/PUT requests
Returns:
The response status and body
"""
try:
req_headers = {"User-Agent": "ZeroClaw/1.0"}
if headers:
for h in headers.split(","):
if ":" in h:
k, v = h.split(":", 1)
req_headers[k.strip()] = v.strip()
data = body.encode() if body else None
req = urllib.request.Request(url, data=data, headers=req_headers, method=method.upper())
with urllib.request.urlopen(req, timeout=30) as resp:
body_text = resp.read().decode("utf-8", errors="replace")
return f"Status: {resp.status}\n{body_text[:5000]}"
except urllib.error.HTTPError as e:
error_body = e.read().decode("utf-8", errors="replace")[:1000]
return f"HTTP Error {e.code}: {error_body}"
except Exception as e:
return f"Error: {e}"
@tool
def web_search(query: str) -> str:
"""
Search the web using Brave Search API.
Requires BRAVE_API_KEY environment variable to be set.
Args:
query: The search query
Returns:
Search results as formatted text
"""
api_key = os.environ.get("BRAVE_API_KEY", "")
if not api_key:
return "Error: BRAVE_API_KEY environment variable not set. Get one at https://brave.com/search/api/"
try:
encoded_query = urllib.parse.quote(query)
url = f"https://api.search.brave.com/res/v1/web/search?q={encoded_query}"
req = urllib.request.Request(
url, headers={"Accept": "application/json", "X-Subscription-Token": api_key}
)
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read().decode())
results = []
for item in data.get("web", {}).get("results", [])[:5]:
title = item.get("title", "No title")
url_link = item.get("url", "")
desc = item.get("description", "")[:200]
results.append(f"- {title}\n {url_link}\n {desc}")
if not results:
return "No results found"
return "\n\n".join(results)
except Exception as e:
return f"Error: {e}"