A tool is a typed function whose docstring guides model calls. Each factory binds tools to a host. mx sets the result budget for the calling model.
/opt/hostedtoolcache/Python/3.12.14/x64/lib/python3.12/site-packages/usearch/__init__.py:131: UserWarning: Will download `usearch_sqlite` binary from GitHub.
warnings.warn("Will download `usearch_sqlite` binary from GitHub.", UserWarning)Resolve a path a tool is only going to read. reading=True is the read-outside allowance.
All read-only tools resolve paths through readable and check(reading=True). The host controls read access outside its roots.
A real folder, a real host, and the tools closed over it. With no Kosha index, search_code is ripgrep and outline is the parser.
root = Path(tempfile.mkdtemp()).resolve()/'proj'
(root/'pkg').mkdir(parents=True)
(root/'pkg'/'sizes.py').write_text('def threshold(n):\n "Half of n."\n return n // 2\n')
(root/'pkg'/'use.py').write_text('from .sizes import threshold\n\ndef budget(): return threshold(8192)\n')
host = LocalHost([root], index=False)
ct = {t.__name__: t for t in code_tools(host)}
sorted(ct)['grep', 'list_files', 'ls', 'outline', 'search_code', 'similar_code']test_eq(sorted(ct), ['grep', 'list_files', 'ls', 'outline', 'search_code', 'similar_code'])
assert 'public_api' not in ct, 'no index, so nothing to list a public surface from'
assert 'threshold' in ct['search_code']('threshold')
assert 'sizes.py' in ct['list_files']('sizes')
assert 'threshold' in ct['outline']('pkg/sizes.py')A no-match result names the search engine. This distinguishes no matches from an unavailable index.
'no matches (Kosha sync in progress; literal fallback via ripgrep)'The fallback above must work before indexing finishes. A second host below builds a real Kosha index over a small package. It checks semantic retrieval, indexed exports and references across files.
/Users/71293/code/personal/orgs/shalya/.venv/lib/python3.13/site-packages/tqdm/auto.py:21: TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html
from .autonotebook import tqdm as notebook_tqdm
parse files from /Users/71293/code/personal/orgs/shalya: 0%| | 0/6 [00:00<?, ?it/s]
parse files from /Users/71293/code/personal/orgs/shalya: 100%|██████████| 6/6 [00:00<00:00, 281.78it/s]{'search_code': <function __main__.code_tools.<locals>.search_code(query: str) -> str>,
'grep': <function __main__.code_tools.<locals>.grep(pattern: str, path_filter: str = '', regex: bool = True, ignore_case: bool = False) -> str>,
'ls': <function __main__.code_tools.<locals>.ls(path: str = '') -> str>,
'similar_code': <function __main__.code_tools.<locals>.similar_code(path: str, line: int = 1) -> str>,
'outline': <function __main__.code_tools.<locals>.outline(path: str) -> str>,
'list_files': <function __main__.code_tools.<locals>.list_files(pattern: str = '') -> str>,
'public_api': <function __main__.code_tools.<locals>.public_api(package: str) -> str>}'[Kosha semantic + keyword index over 1 folder(s) and environment fused with ripgrep]\n/Users/71293/code/personal/orgs/shalya/shalya/tools.py:785 shalya.tools.tools_for def tools_for(host, get_skills=None, extra=(), mx=MAX_TOOL_CHARS, drop=(), image=None): """Every tool this host declares it can support, plus whatever else was registered. `image` is a built image group, or None. It needs to know what the t\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/shalya/nbs/02_tools.ipynb:524 "\'[Kosha semantic + keyword index over 1 folder(s) and environment fused with ripgrep]\\\\n/Users/71293/code/personal/orgs/shalya/nbs/02_tools.ipynb:473 \\"\\\\\'[Kosha semantic + keyword index over 1 fo\n NOTEBOOK -- use this exact path with notebook_cells, then view_cell/edit_cell\n/Users/71293/code/personal/orgs/rishi/.venv/lib/python3.13/site-packages/matplotlib/backend_tools.py:982 matplotlib.backend_tools.add_tools_to_container def add_tools_to_container(container, tools=None): """ Add multiple tools to the container. Parameters ---------- container : Container `.backend_bases.ToolContainerBase` object that will get the tools added. tools : list, optional List in \n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/shalya/nbs/02_tools.ipynb:533 "indexed_tools[\'search_code\'](\'code tools\')"\n NOTEBOOK -- use this exact path with notebook_cells, then view_cell/edit_cell\n/Users/71293/Library/Application Support/uv/tools/koshas/lib/python3.13/site-packages/mcp/server/mcpserver/tools/tool_manager.py:19 mcp.server.mcpserver.tools.tool_manager.ToolManager class ToolManager: """Manages MCPServer tools.""" def __init__(self, warn_on_duplicate_tools: bool = True, *, tools: list[Tool] | None = None): self._tools: dict[str, Tool] = {} for tool in tools or (): if warn_on_duplicate_tools and tool.n\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/shalya/shalya/skills.py:161 shalya.skills.Registry.tool def tool(self, f): "Register a tool." self.tools.append(f) return f\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/rishi/.venv/lib/python3.13/site-packages/mlx_audio/tts/models/qwen3_tts/qwen3_tts.py:957 mlx_audio.tts.models.qwen3_tts.qwen3_tts.Model._predict_code_tokens def _predict_code_tokens( self, first_token: mx.array, hidden: mx.array, *, temperature: float, top_k: int, top_p: float, code_cache=None, ) -> Tuple[List[mx.array], mx.array]: if code_cache is None: code_cache = self.talker.code_predictor.\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/shalya/shalya/core.py:38 shalya.core.err def err(what, e=None): "One tool failure, spelled the way every other tool spells it." return f\'{ERR}{what}\' + (f\': {host_err(e)}\' if e is not None else \'\')\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/executing/executing.py:731 executing.executing.SentinelNodeFinder.compile_instructions def compile_instructions(self): # type: () -> List[EnhancedInstruction] module_code = compile_similar_to(self.tree, self.code) code = only(self.find_codes(module_code)) return self.clean_instructions(code)\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/rishi/.venv/lib/python3.13/site-packages/matplotlib/backend_bases.py:3478 matplotlib.backend_bases.ToolContainerBase.add_tool def add_tool(self, tool, group, position=-1): """ Add a tool to this container. Parameters ---------- tool : tool_like The tool to add, see `.ToolManager.get_tool`. group : str The name of the group to add this tool to. position : int, defa\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/spacy/cli/_util.py:186 spacy.cli._util.import_code def import_code(code_path: Optional[Union[Path, str]]) -> None: """Helper to import Python file provided in training commands / commands using the config. This makes custom registered functions available. """ if code_path is not None: if no\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/debugpy/_vendored/pydevd/_pydevd_frame_eval/vendored/bytecode/concrete.py:170 debugpy._vendored.pydevd._pydevd_frame_eval.vendored.bytecode.concrete.ConcreteBytecode.from_code def from_code(code, *, extended_arg=False): line_starts = dict(entry for entry in dis.findlinestarts(code) if entry[1] is not None) # find block starts instructions = [] offset = 0 lineno = code.co_firstlineno while offset < (len(code.co_co\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/rich/control.py:58 rich.control.Control.__init__ def __init__(self, *codes: Union[ControlType, ControlCode]) -> None: control_codes: List[ControlCode] = [ (code,) if isinstance(code, ControlType) else code for code in codes ] _format_map = CONTROL_CODES_FORMAT rendered_codes = "".join( _f\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/rishi/.venv/lib/python3.13/site-packages/matplotlib/backend_managers.py:215 matplotlib.backend_managers.ToolManager.add_tool def add_tool(self, name, tool, *args, **kwargs): """ Add *tool* to `ToolManager`. If successful, adds a new event ``tool_trigger_{name}`` where ``{name}`` is the *name* of the tool; the event is fired every time the tool is triggered. Param\n FILE -- use this exact path with view_file/edit_file\n/Users/71293/code/personal/orgs/rishi/.venv/lib/python3.13/site-packages/transformers/models/xcodec2/modular_xcodec2.py:244 transformers.models.xcodec2.modular_xcodec2.Xcodec2DecoderLayer class Xcodec2DecoderLayer(LlamaDecoderLayer): pass\n FILE -- use this exact path with view_file/edit_file…[truncated: 5939 of 7996 chars shown]'"125 public name(s) in shalya.host\nshalya.host.ApiHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:311 Reading an API specification and calling what it describes.\nshalya.host.AskHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:184 Model-backed answers from memory.\nshalya.host.Capability /Users/71293/code/personal/orgs/shalya/shalya/host.py:21 One capability group. A host declares the group by inheriting the class that names it.\nshalya.host.CodeHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:87 Code search and structure.\nshalya.host.GitHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:312 Git operations on a working tree inside `roots`.\nshalya.host.Host /Users/71293/code/personal/orgs/shalya/shalya/host.py:27 The application under an agent: the folders it may touch, and what it declares it can do.\nshalya.host.LocalHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:403 Reference host for local folders.\nshalya.host.MemoryHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:155 Durable memory organized as document sections.\nshalya.host.NotebookHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:142 Notebook cell listing and insertion.\nshalya.host.SessionHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:239 The live namespace, and the terminal beside it.\nshalya.host.ShellHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:288 Running a command on the machine.\nshalya.host.WatchHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:193 Recurring watches and reminders.\nshalya.host.WebHost /Users/71293/code/personal/orgs/shalya/shalya/host.py:122 Current web search and page retrieval.\nshalya.host._defs /Users/71293/code/personal/orgs/shalya/shalya/host.py:757 Every def/class in one file as `(line, qualified_name, depth)`, by parsing rather than grepping.\nshalya.host._exec /Users/71293/code/personal/orgs/shalya/shalya/host.py:783 Run `code` in `ns`, returning printed output plus the last expression's value.\nshalya.host._fossick /Users/71293/code/personal/orgs/shalya/shalya/host.py:910 \nshalya.host._ranked /Users/71293/code/personal/orgs/shalya/shalya/host.py:626 Call Kosha with optional cross-encoder reranking.\nshalya.host._rg /Users/71293/code/personal/orgs/shalya/shalya/host.py:591 Search with rgapi; `every_file=True` matches `walk`.\nshalya.host._scan /Users/71293/code/personal/orgs/shalya/shalya/host.py:666 Find literal matches by reading files.\nshalya.host._semantic /Users/71293/code/personal/orgs/shalya/shalya/host.py:637 Kosha hybrid results (repo + env + graph) as the Host's stable `Hit` shape.\nshalya.host._walk /Users/71293/code/personal/orgs/shalya/shalya/host.py:478 Files under `root`, skipping the same generated dirs/suffixes `grep` covers.\nshalya.host.add_root /Users/71293/code/personal/orgs/shalya/shalya/host.py:442 return resolved folders for read/write. does not create folders\nshalya.host.added_roots /Users/71293/code/personal/orgs/shalya/shalya/host.py:462 Roots opened after construction. `/resume` lapses these, and says that it did.\nshalya.host.api_call /Users/71293/code/personal/orgs/shalya/shalya/host.py:1060 \nshalya.host.api_count /Users/71293/code/personal/orgs/shalya/shalya/host.py:1054 \nshalya.host.api_load /Users/71293/code/personal/orgs/shalya/shalya/host.py:1042 \nshalya.host.api_ops /Users/71293/code/personal/orgs/shalya/shalya/host.py:1048 \nshalya.host.approvals /Users/71293/code/personal/orgs/shalya/shalya/host.py:566 \nshalya.host.ask /Users/71293/code/personal/orgs/shalya/shalya/host.py:1012 \nshalya.host.check /Users/71293/code/personal/orgs/shalya/shalya/host.py:456 Resolve `path`. Refuse outside `roots` (unless `read_outside` and `reading`). Walks stay confined.\nshalya.host.denied /Users/71293/code/personal/orgs/shalya/shalya/host.py:361 Whether `path` matches a refused credential path.\nshalya.host.grep /Users/71293/code/personal/orgs/shalya/shalya/host.py:654 Exact matching through ripgrep. None when `rgapi` is unavailable. The tool then reads files itself.\nshalya.host.implemented /Users/71293/code/personal/orgs/shalya/shalya/host.py:1018 Recompute what `cls` is still missing, after `@patch` filled some of it in.\nshalya.host.index_ready /Users/71293/code/personal/orgs/shalya/shalya/host.py:572 Whether all open roots are indexed.\nshalya.host.indexed /Users/71293/code/personal/orgs/shalya/shalya/host.py:578 Indexed root paths.\nshalya.host.inspect_python /Users/71293/code/personal/orgs/shalya/shalya/host.py:842 Run `code` without rebinding anything the user made. A kernel enforces this; a copy approximates it.\nshalya.host.kernel_kind /Users/71293/code/personal/orgs/shalya/shalya/host.py:817 \nshalya.host.ld_json /Users/71293/code/personal/orgs/shalya/shalya/host.py:394 Parse `schema.org` JSON-LD blocks from `html`.\nshalya.host.list_vars /Users/71293/code/personal/orgs/shalya/shalya/host.py:862 \nshalya.host.memory_forget /Users/71293/code/personal/orgs/shalya/shalya/host.py:1000 \nshalya.host.memory_read /Users/71293/code/personal/orgs/shalya/shalya/host.py:988 \nshalya.host.memory_search /Users/71293/code/personal/orgs/shalya/shalya/host.py:976 \nshalya.host.memory_topics /Users/71293/code/personal/orgs/shalya/shalya/host.py:947 \nshalya.host.memory_tree /Users/71293/code/personal/orgs/shalya/shalya/host.py:935 \nshalya.host.nb_add_cell /Users/71293/code/personal/orgs/shalya/shalya/host.py:806 \nshalya.host.nb_cells /Users/71293/code/personal/orgs/shalya/shalya/host.py:799 \nshalya.host.note /Users/71293/code/personal/orgs/shalya/shalya/host.py:570 \nshalya.host.peers /Users/71293/code/personal/orgs/shalya/shalya/host.py:751 Find references to the symbol defined at `path`:`line`.\nshalya.host.poll /Users/71293/code/personal/orgs/shalya/shalya/host.py:1036 \n…[76 more line(s) not shown. read one with view_file]"'/Users/71293/code/personal/orgs/shalya/shalya/host.py:27 shalya.host.Host class Host(ABC): "The application under an agent: the folders it may touch, and what it declares it can do." group = \'file\' #: every host has the path boundary the file tools need without = frozenset() #: groups this instance cannot do, wha\n/Users/71293/code/personal/orgs/shalya/shalya/core.py:34 shalya.core.host_err def host_err(e): "A caught exception, for a user-facing surface." return f\'{type(e).__name__}: {e}\'\n/Users/71293/code/personal/orgs/shalya/README.md:16 from shalya.host import LocalHost\n/Users/71293/code/personal/orgs/shalya/shalya/tools.py:785 shalya.tools.tools_for def tools_for(host, get_skills=None, extra=(), mx=MAX_TOOL_CHARS, drop=(), image=None): """Every tool this host declares it can support, plus whatever else was registered. `image` is a built image group, or None. It needs to know what the t\n/Users/71293/code/personal/orgs/shalya/README.md:23 [`LocalHost`](https://vedicreader.github.io/shalya/host.html#localhost) touches only the folders you open, and reports the capability groups it can serve. A group with no installed backend goes absent\n/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/mlx/_distributed_utils/config.py:434 _distributed_utils.config.prepare_ethernet_hostfile def prepare_ethernet_hostfile(args, hosts): log(args.verbose, f"Preparing an ethernet hostfile") add_ips(hosts, args.verbose) hostfile = Hostfile( [Host(i, h.ssh_hostname, h.ips, []) for i, h in enumerate(hosts)], "", args.env ) save_hostfi\n/Users/71293/code/personal/orgs/shalya/README.md:33 host = LocalHost(roots=[d], index=False, web=False)\n/Users/71293/code/personal/orgs/shalya/shalya/core.py:33 shalya.core.HostError class HostError(Exception): "A host refusal."\n/Users/71293/code/personal/orgs/shalya/tests/test_summaries.py:15 from shalya.host import LocalHost\n/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/fasthtml/core.py:643 fasthtml.core.HostRoute class HostRoute(Route): "Route with optional host-header constraint using Starlette\'s {param} pattern syntax" def __init__(self, path, endpoint, *, host=None, **kwargs): super().__init__(path, endpoint, **kwargs) self.host = host if host: s\n/Users/71293/code/personal/orgs/shalya/tests/test_summaries.py:21 class AnyHost:\n/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/curl_cffi/requests/cookies.py:171 curl_cffi.requests.cookies.Cookies._eff_request_host def _eff_request_host(self, request) -> str: """ Almost equivalent to the eff_request_host function in: https://github.com/python/cpython/blob/3.11/Lib/http/cookiejar.py#L636 """ host = urlparse(request.url)[1] if host == "": host = request\n/Users/71293/code/personal/orgs/shalya/tests/test_summaries.py:31 h = AnyHost()\n/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/pydantic/v1/networks.py:439 pydantic.v1.networks.MultiHostDsn._build_url def _build_url(cls, m: Match[str], url: str, parts: \'Parts\') -> \'MultiHostDsn\': hosts_parts: List[\'HostParts\'] = [] host_re = host_regex() for host in m.groupdict()[\'hosts\'].split(\',\'): d: Parts = host_re.match(host).groupdict() # type: ign\n/Users/71293/code/personal/orgs/shalya/tests/test_summaries.py:32 out = list(code_tools(LocalHost([tmp_path], index=False, web=False)))\n/Users/71293/code/personal/orgs/leela/.venv/lib/python3.13/site-packages/huggingface_hub/_sandbox.py:1625 huggingface_hub._sandbox.SandboxPool._save_cache def _save_cache(self) -> None: """Persist the pool config + current hosts (with their live counts) for next time.""" with self._lock: hosts = [ CachedHost( job_id=host.job_id, owner=host.owner, base_url=host.base_url, nonce=host.nonce, capa\n/Users/71293/code/personal/orgs/shalya/nbs/01_host.ipynb:59 "from shalya.core import Hit, HostError, MAX_API, MAX_FILE, MAX_GREP_HITS, host_err"\n/Users/71293/Library/Application Support/uv/tools/koshas/lib/python3.13/site-packages/mcp/server/transport_security.py:50 mcp.server.transport_security.TransportSecurityMiddleware._validate_host def _validate_host(self, host: str | None) -> bool: """Validate the Host header against allowed values.""" if not host: logger.warning("Missing Host header in request") return False # Check exact match first if host in self.settings.allowed\n/Users/71293/code/personal/orgs/shalya/nbs/01_host.ipynb:94 "The path boundary is not a capability. Every group needs it, so it is `Host` itself."\n/Users/71293/code/personal/orgs/kosha/.venv/lib/python3.13/site-packages/jupyter_server/base/handlers.py:556 jupyter_server.base.handlers.JupyterHandler.check_host def check_host(self) -> bool: """Check the host header if remote access disallowed. Returns True if the request should continue, False otherwise. """ if self.settings.get("allow_remote_access", False): return True # Remove port (e.g. \':8888'view_file returns lineno|hash|content. Edits must quote a current address. A stale hash is refused, preventing edits to unseen content.
Reading and editing files, by exact text or by hash-verified address.
All commands succeed or nothing is written.
1|2337|def greet(name):
2|709e| return "hello " + namereplace_text is the same write through the exact-text vocabulary, and it refuses for the same reasons apply_edits refuses.
Every write tool carries its own mark, and the frozenset of names says the same thing for a caller that only has a name.
Notebooks, addressed by cell id rather than by line.
The two operations that need to know what a notebook is, and the two that only need a cell id.
('added cell 75ea04d7 to nb/demo.ipynb', '75ea04d7 code x = 1')assert cid in nt['notebook_cells']('nb/demo.ipynb')
assert 'x = 1' in nt['view_cell']('nb/demo.ipynb', cid)
line = nt['view_cell']('nb/demo.ipynb', cid).splitlines()[0]
nt['edit_cell']('nb/demo.ipynb', cid, json.dumps([['|'.join(line.split('|')[:2]) + '|', 's', 'x = 1', 'x = 2']]))
assert 'x = 2' in nt['view_cell']('nb/demo.ipynb', cid), nt['view_cell']('nb/demo.ipynb', cid)Two groups, and the split is the point. The web tools go out now. The memory tools recall what going out already found, without going out again.
The web, for the questions whose answer depends on current documentation.
A result provider isolates the tool contract. These checks cover model-facing text and argument forwarding; host covers fossick itself.
class WebResults:
research_note = 'fixture'
def web_search(self, query, n=20):
return [AttrDict(title='Python', url='https://python.org')] if query else []
def read_url(self, url, remember=True):
self.remember = remember
return AttrDict(text='# Python', url=url)
def research(self, query): return 'Python is a programming language.'
web_results = WebResults()
wt = {t.__name__: t for t in web_tools(web_results)}test_eq(sorted(wt), ['read_url', 'research', 'web_search'])
test_eq(wt['web_search']('python'), 'Python\n https://python.org')
test_eq(wt['web_search'](''), 'no results (fixture)')
test_eq(wt['read_url']('https://python.org', remember=False), '# Python')
test_eq(web_results.remember, False)
test_eq(wt['research']('python'), 'Python is a programming language.')
assert not any(is_write(t) for t in wt.values())Durable pages and research recalled as document sections rather than flat snippets.
Standing interests: what to put back on the desk later, and what has come due now.
A watch is registered, listed, polled and cancelled, and a reminder is a watch that files its own text.
Answering out of memory is its own group. It takes a model, and a host can remember pages and search them without having anything to ask them with.
A deterministic result provider covers JSON rendering, citations, watch summaries, errors and write marks. host covers storage adapters.
class MemoryResults:
def memory_search(self, query, limit=8):
if query == 'fail': raise RuntimeError('vault unavailable')
return [{'id': 'd1', 'title': 'Kettles', 'text': 'boils at 100C'}][:limit]
def memory_tree(self, document=''): return [{'id': 'd1', 'title': 'Kettles'}]
def memory_read(self, node_id): return {'id': node_id, 'text': 'boils at 100C'}
def memory_topics(self, limit=12): return [{'topic': 'home'}][:limit]
def memory_forget(self, doc_id): return doc_id == 'd1'
def remember(self, text, title=None, tags=()): return {'doc_id': 'd2', 'title': title or text}
def watch(self, target, action='remind', every='1d', note=None):
return {'id': 'w1', 'target': target, 'action': action, 'every': 86400, 'runs': 0}
def watches(self, due_only=False):
return [] if due_only else [{'id': 'w1', 'target': 'renew domain', 'action': 'remind',
'every': 86400, 'runs': 0, 'last_status': None}]
def unwatch(self, watch_id): return watch_id == 'w1'
def poll(self):
return {'ran': 1, 'checked': 1,
'results': [{'status': 'ok', 'action': 'remind', 'target': 'renew domain'}]}
def ask(self, question, ref=None, instruction=''):
return {'answer': 'Water boils at 100C.',
'cited': [{'n': 1, 'breadcrumb': 'Kettles', 'node_id': 'd1'}]}
memory_results = MemoryResults()
mt = {t.__name__: t for t in memory_tools(memory_results)}
wa = {t.__name__: t for t in watch_tools(memory_results)}
at = {t.__name__: t for t in ask_tools(memory_results)}test_eq(sorted(mt), ['memory_forget', 'memory_read', 'memory_search', 'memory_topics', 'memory_tree'])
test_eq(sorted(wa), ['cancel_watch', 'list_watches', 'poll_watches', 'remember', 'set_reminder', 'watch_url'])
test_eq(sorted(at), ['ask_memory'])
assert 'boils at 100C' in mt['memory_search']('kettle')
assert failed(mt['memory_search']('fail')) and 'vault unavailable' in mt['memory_search']('fail')
test_eq(mt['memory_forget']('d1'), 'forgot document')
test_eq(wa['remember']('fact', title='Facts', tags='one, two'), "remembered 'Facts' as d2")
test_eq(wa['set_reminder']('renew domain'), 'reminder w1 set, every 1w')
assert 'every 86400s' in wa['list_watches']() and 'renew domain' in wa['list_watches']()
test_eq(wa['poll_watches'](), '1 of 1 fired\nok remind renew domain')
test_eq(wa['cancel_watch']('w1'), 'cancelled w1')
answer = at['ask_memory']('boiling point')
assert 'Water boils at 100C.' in answer and '[1] Kettles (d1)' in answer
test_eq({t.__name__ for t in mt.values() if is_write(t)}, {'memory_forget'})
test_eq({t.__name__ for t in wa.values() if is_write(t)}, {'cancel_watch'})The live kernel the user is working in, and the terminal they are looking at.
The session tools reach the live namespace, and the shell tool reaches the machine. Both are write tools, and both come back as text rather than raising.
('(no output)', '42', 'exit 0\nhi')test_eq(st['run_python']('total'), '42')
assert 'total' in st['list_vars']()
assert 'hi' in sh['run_shell']('echo hi')
host.note('a status line')
assert 'a status line' in st['read_terminal'](), st['read_terminal']()
assert is_write(st['run_python']) and is_write(sh['run_shell'])
assert not is_write(st['inspect_python']) and not is_write(st['read_terminal'])
test_eq(sorted(st), ['inspect_python', 'list_vars', 'read_terminal', 'run_python'])Read an API specification, browse what it declares, and call one operation.
A deterministic result provider covers JSON rendering, pagination, parameter forwarding and errors. host covers API backends.
class ApiResults:
def api_load(self, src, name=''): return {'name': name or 'petstore', 'operations': 3}
def api_ops(self, group='', name='', match='', offset=0):
rows = [{'name': 'listPets'}, {'name': 'getPet'}, {'name': 'createPet'}]
if match: rows = [r for r in rows if match.lower() in r['name'].lower()]
return rows[offset:offset + 1]
def api_count(self, group='', name='', match=''):
return len([n for n in ('listPets', 'getPet', 'createPet') if not match or match.lower() in n.lower()])
def api_call(self, operation, name='', **params):
if operation == 'fail': raise RuntimeError('service unavailable')
return {'called': operation, 'with': params}
api_results = ApiResults()
apit = {t.__name__: t for t in api_tools(api_results)}test_eq(sorted(apit), ['api_call', 'api_load', 'api_ops'])
test_eq(json.loads(apit['api_load']('spec.json')), {'name': 'petstore', 'operations': 3})
page = json.loads(apit['api_ops']())
test_eq(page['operations'], [{'name': 'listPets'}])
test_eq((page['matched'], page['showing'], page['more']), (3, '1-1', 'call again with offset=1'))
test_eq(json.loads(apit['api_ops'](match='get'))['operations'], [{'name': 'getPet'}])
test_eq(json.loads(apit['api_call']('getPet', params={'id': 7})),
{'called': 'getPet', 'with': {'id': 7}})
assert failed(apit['api_call']('fail')) and 'service unavailable' in apit['api_call']('fail')Reading discovered skills and creating project-local Agent Skills.
The system prompt lists skill names and descriptions. read_skill fetches a body when the agent needs it.
skill_path = root/'.agents'/'skills'/'house-style'/'SKILL.md'
skill_path.parent.mkdir(parents=True)
skill_path.write_text(
'---\nname: house-style\ndescription: How code is written here.\n---\n\nDense lines. Short names.\n')
skills = discover(roots=[root])
skt = {t.__name__: t for t in skill_tools(host, lambda: skills)}
skt['read_skill']('house-style')'<skill name="house-style" from="/private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp0nwf2kip/proj/.agents/skills/house-style/SKILL.md">\n\nDense lines. Short names.\n\n</skill>'test_eq(sorted(skt), ['create_skill', 'read_skill'])
body = skt['read_skill']('house-style')
assert body.startswith('<skill name="house-style"') and 'Dense lines.' in body
test_eq(skt['read_skill']('house'), body) # unique prefix resolves to the same skill
assert "no skill matching 'nope'" in skt['read_skill']('nope')
assert 'house-style' in skt['read_skill']('nope')
test_eq({t.__name__ for t in skill_tools(host, lambda: skills) if is_write(t)}, {'create_skill'})create_skill writes under the first open folder in the layout discover reads, and refuses rather than overwriting one that is already there.
'created /private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp0nwf2kip/proj/.agents/skills/notebook-tests/SKILL.md; run /reload to load it into the current agent'made = root/'.agents'/'skills'/'notebook-tests'/'SKILL.md'
assert made.exists(), made
assert 'when testing' in made.read_text() and 'readable case' in made.read_text()
assert skt['create_skill']('notebook-tests', 'again', 'body').startswith('refusing to overwrite')
assert 'kebab-case' in skt['create_skill']('Notebook Tests', 'x', 'y')Image generation uses the current model when it can draw, then a dedicated image endpoint. draws_itself and from_reply provide the model-specific operations.
Save one media item under the session and return its path.
save_media chooses the path and name for a generated picture. A frontend can locate the result from its session.
(Path('/private/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/tmp0nwf2kip/proj/s1/media/generated-1.png'),
'image/png')assert saved.exists() and saved.suffix == '.png'
test_eq(saved.parent, media_dir(root/'s1'))
test_eq(saved.read_bytes(), png['data'])
test_eq(mime_for(saved), 'image/png')
assert saved.name.startswith('generated'), saved.name
assert save_media(png, session=root/'s1', stem='generated').name != saved.nameDrawing: by the turn’s own model where it can, and by the images endpoint where it cannot.
The image group is credentialled rather than declared: a key is either there or it is not, and tools_for never probes for one. Without a key the tool says so and reaches nothing.
(True, 'gpt-image-1', 'gpt-image-1')test_eq(image_available(), bool(os.environ.get('OPENAI_API_KEY')))
test_eq(api_model('openai/gpt-image-1'), 'gpt-image-1') # a vendor prefix is stripped
test_eq(api_model('gpt-image-1'), 'gpt-image-1')
assert '1024x1024' in IMAGE_SIZES and 'auto' in IMAGE_SIZES
test_eq(IMAGE_MODEL, 'gpt-image-1')
assert RESPONSES_API.startswith('https://') and IMAGE_API.startswith('https://')
assert all(v.endswith('/') for v in API_VENDORS)
gen = image_tools(host)[0]
test_eq(gen.__name__, 'generate_image')
if not image_available():
assert 'OPENAI_API_KEY' in gen('a kettle') # it says so rather than reaching the wire
else: assert failed(gen('a kettle', size='3x3')) # and an unknown size never gets sentThe live image check calls the configured OpenAI image endpoint and verifies that the returned file is a non-empty image. It is excluded from the default suite because it spends credentials.
image_root = Path(tempfile.mkdtemp())
image_host = LocalHost([image_root], index=False)
generate_image = image_tools(image_host, session=image_root)[0]
image_result = generate_image('A simple black circle centered on a white background', n=1)
assert not image_result.startswith('ERROR:'), image_result
image_path = Path(image_result.strip())
assert image_path.exists() and image_path.stat().st_size > 1000
assert mime_for(image_path).startswith('image/')Git tools delegate to gheasy. A repository found by walking above a supplied path must still be inside the host’s roots.
Git bound to one open repository, kept inside the host’s roots.
A real repository, with one commit and one untracked file.
repo = Path(tempfile.mkdtemp()).resolve()/'repo'
repo.mkdir(parents=True)
(repo/'a.py').write_text('x = 1\n')
for c in (['init', '-q', '-b', 'main'], ['add', 'a.py'],
['-c', 'user.email=t@e', '-c', 'user.name=T', 'commit', '-qm', 'first']):
subprocess.run(['git', '-C', str(repo)] + c, check=True)
(repo/'b.py').write_text('y = 2\n')
ghost = LocalHost([repo], index=False)
gt = {t.__name__: t for t in git_tools(ghost)}
status = json.loads(gt['git_status']())
status['branch'], status['clean'], [c['path'] for c in status['changes']]('main', False, ['b.py'])test_eq(status['branch'], 'main')
test_eq(status['clean'], False)
test_eq([c['path'] for c in status['changes']], ['b.py'])
test_eq(sorted(gt), sorted(GIT_TOOLS))
test_eq({t.__name__ for t in git_tools(ghost) if is_write(t)}, set(GIT_WRITE_TOOLS))
assert failed(gt['git_divergence']()) and 'tracks nothing' in gt['git_divergence']()
prev = json.loads(gt['git_rebase_preview']('main'))
assert {'current', 'onto', 'clean', 'merge_base'} <= set(prev), sorted(prev)
assert 'stops_at' in prev and 'replayed' in prev, sorted(prev)
test_eq(subprocess.run(['git', '-C', str(repo), 'status', '--porcelain'],
capture_output=True, text=True).stdout.strip(), '?? b.py')
assert failed(gt['git_checkout']('no-such-branch'))
assert 'op must be one of' in gt['git_remote']('nonsense') # never reaches the network
fetched = gt['git_remote']('fetch') # no remote, so it is a no-op not a lie
assert failed(fetched) or json.loads(fetched)['branch'] == 'main', fetchedHost.provides selects tool groups. drop removes groups from that selection.
Every tool this host declares it can support, plus whatever else was registered.
image is a built image group, or None. It needs to know what the turn’s model can do, and a host does not, so the frontend that knows builds it and passes it in.
tools_for maps declared group names to factories. Tests require the table and capability classes to use the same names.
['code',
'file',
'notebook',
'web',
'memory',
'ask',
'watch',
'api',
'session',
'shell',
'git']from shalya.host import (ApiHost, AskHost, CodeHost, GitHost, MemoryHost, NotebookHost,
SessionHost, ShellHost, WatchHost, WebHost)
declared = {c.group for c in (CodeHost, WebHost, NotebookHost, MemoryHost, AskHost, WatchHost,
SessionHost, ShellHost, ApiHost, GitHost)} | {'file'}
test_eq({g for g, _ in GROUPS}, declared)
test_eq(len(GROUPS), len(declared)) # named once each
assert all(callable(f) for _, f in GROUPS)A host over a git repository gets ten groups’ worth of tools, and the groups it declared it could not do are simply absent.
['add_cell',
'add_root',
'create_file',
'edit_cell',
'edit_file',
'git_checkout',
'git_divergence',
'git_rebase_preview',
'git_remote',
'git_status',
'grep',
'inspect_python',
'list_files',
'list_vars',
'ls',
'notebook_cells',
'outline',
'read_terminal',
'read_url',
'replace_text',
'research',
'run_python',
'run_shell',
'search_code',
'similar_code',
'view_cell',
'view_file',
'web_search']file is not a group a host declares. Every host has the path boundary, so every host gets the file tools, and the table names it so a budget can still drop it.
A sub-agent gets the read tools and, with a budget, a hard cap on how many calls it may make. The write set comes from the tools themselves: is_write reads the mark @writes left.
The tools an agent may have when it must not act, optionally behind a hard call budget.
Every write tool is filtered out, whatever else the host offers. effects=False withholds the rest of the ways to act: running code, an API call that can POST, an image that costs money, and standing work that outlives the turn. None of those writes a file the user owns, so WRITE_TOOLS does not name them and approval does not gate them. An agent asked to propose and not to act is refused them all the same. What survives can still be pinned: read_url keeps its page and loses the vault entry.
['search_code',
'grep',
'ls',
'similar_code',
'outline',
'list_files',
'view_file',
'notebook_cells',
'view_cell',
'web_search',
'read_url',
'research',
'list_vars',
'inspect_python',
'read_terminal',
'git_status',
'git_divergence',
'git_rebase_preview']ro = {t.__name__ for t in read_only(ts)}
test_eq(ro & WRITE_TOOLS, set())
assert 'search_code' in ro and 'git_status' in ro
test_eq({t.__name__ for t in read_only(ts, writes=True)} & {'create_file'}, {'create_file'})
safe_web = {t.__name__: t for t in read_only(wt.values())}
test_eq(wt['read_url'].read_only, safe_web['read_url'])
safe_web['read_url']('https://python.org')
test_eq(web_results.remember, False)
test_eq(set(get_schema(safe_web['read_url'])['input_schema']['properties']), {'url'})
full_web = {t.__name__: t for t in read_only(wt.values(), writes=True)}
full_web['read_url']('https://python.org')
test_eq(web_results.remember, True)
# `effects=False` also withholds what acts without writing a file the user owns
acting = {t.__name__ for t in ts if has_effect(t)}
test_eq(acting - ACTING_TOOLS, set()) # nothing is marked without being named
assert acting <= ro # and the default keeps every one of them
propose = {t.__name__ for t in read_only(ts, effects=False)}
test_eq(propose & (WRITE_TOOLS | ACTING_TOOLS), set())
assert 'search_code' in propose # looking is still allowedWith a budget, the tools themselves stop the loop. A local engine owns its internal tool loop, so a cap the harness holds is a cap the harness cannot enforce.
('no matches (Kosha sync in progress; literal fallback via ripgrep)',
'Tool budget exhausted. Stop calling tools and return the best evidence-backed answer now.')