kosha

kosha

Find the code you need before you write it.

kosha (कोश, “treasury”) keeps a searchable index of your repository and of the Python packages it depends on. Ask it in plain words, such as “atomic write to a temp file”, and it returns the functions and classes that do that, from your code and from your dependencies, with their file, line and source. Ask it about a function and it tells you what calls it, what it calls, and how much code depends on it, so you can judge what a change will touch.

It is built for coding agents as much as for people. An agent that searches before it writes reuses the helper your project already has instead of adding a second one. kosha runs locally, uses no language model, and returns real code you can read.

Three ideas recur on this page. A chunk is one function, class or method cut out of a source file and stored with its metadata. The index finds chunks in two ways at once: keyword search, which matches the words of the query, and vector search, which compares embeddings, lists of numbers computed from text so that code with similar meaning gets similar numbers. The call graph records which function uses which, across your code and your packages.

Install

kosha is a development tool, so add it as a dev dependency. The package is published as koshas; the module you import and the command you run are both kosha.

uv add --dev koshas

kosha ships a skill file, SKILL.md, that tells coding agents how to use it. Install it once per project:

Kosha(install_skill=True)   # writes .agents/skills/kosha/SKILL.md and .claude/skills/kosha/SKILL.md

Commit .agents/skills/kosha/SKILL.md so that every contributor’s agent picks it up.

Build the index

Kosha() opens the index for the repository you are in. sync() brings it up to date: it indexes your repo files, the packages listed in pyproject.toml with their direct dependencies, and the call graph between them. The first sync takes a while. Later syncs re-read only files and package versions that changed.

k = Kosha()
k.sync()

k.sync(graph=False) skips the call graph when you only need search. graph_mode='full' re-extracts the graph in worker processes. graph_metrics=False skips recomputing PageRank (explained below); call k.graph.recompute_metrics() once you are done updating. The example below syncs only two packages.

k = Kosha()
k.sync(pkgs=['fastcore', 'litesearch'])

/Users/71293/code/personal/orgs/kosha/.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

Syncing dir=/Users/71293/code/personal/orgs/kosha, repo=True, env=True, graph=True, force=False
loading pkgs ['fastcore', 'litesearch'] ...

Updating packages:   0%|                                                                                                               | 0/2 [00:00<?, ?pkg/s]

updating pkg: fastcore ...

<style> progress { appearance: none; border: none; border-radius: 4px; width: 300px; height: 20px; vertical-align: middle; background: #e0e0e0; } progress::-webkit-progress-bar { background: #e0e0e0; border-radius: 4px; } progress::-webkit-progress-value { background: #2196F3; border-radius: 4px; } progress::-moz-progress-bar { background: #2196F3; border-radius: 4px; } progress:not([value]) { background: repeating-linear-gradient(45deg, #7e7e7e, #7e7e7e 10px, #5c5c5c 10px, #5c5c5c 20px); } progress.progress-bar-interrupted::-webkit-progress-value { background: #F44336; } progress.progress-bar-interrupted::-moz-progress-value { background: #F44336; } progress.progress-bar-interrupted::-webkit-progress-bar { background: #F44336; } progress.progress-bar-interrupted::-moz-progress-bar { background: #F44336; } progress.progress-bar-interrupted { background: #F44336; } table.fastprogress { border-collapse: collapse; margin: 1em 0; font-size: 0.9em; } table.fastprogress th, table.fastprogress td { padding: 8px 12px; border: 1px solid #ddd; text-align: left; } table.fastprogress thead tr { background: #f8f9fa; font-weight: bold; } table.fastprogress tbody tr:nth-of-type(even) { background: #f8f9fa; } </style>

syncing files [Path('/Users/71293/code/personal/orgs/kosha/kosha/skill.py')] .....


parse files from /Users/71293/code/personal/orgs/kosha:   0%|                                                                           | 0/1 [00:00<?, ?it/s]parse files from /Users/71293/code/personal/orgs/kosha: 100%|█████████████████████████████████████████████████████████████████| 1/1 [00:00<00:00, 1310.72it/s]

loading code graph for packages:   0%|                                                                                                 | 0/2 [00:00<?, ?pkg/s]

{'changed': 0, 'same': 0, 'removed': 0}
synced repo

loading code graph for packages: 100%|████████████████████████████████████████████████████████████████████████████████████████| 2/2 [00:00<00:00, 133.39pkg/s]
Updating packages:  50%|███████████████████████████████████████████████████▌                                                   | 1/2 [00:00<00:00,  8.73pkg/s]

package {'name': 'fastcore', 'version': '2.2.16'} already loaded.
updating pkg: litesearch ...

Updating packages: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████| 2/2 [00:00<00:00,  8.54pkg/s]Updating packages: 100%|███████████████████████████████████████████████████████████████████████████████████████████████████████| 2/2 [00:00<00:00,  8.56pkg/s]

package {'name': 'litesearch', 'version': '0.1.32'} already loaded.

[None, None, <kosha.graph.CodeGraph object>]

k.status()

{'files': 5,
 'packages': 498,
 'graph_nodes': 10623,
 'stale_files': 0,
 'stale_pkgs': {},
 'new_files': 1}

status tells you whether the index is current. stale_files counts repo files changed since they were indexed, new_files counts Python files not indexed yet, and stale_pkgs lists installed packages not yet indexed at their installed version. Run k.sync() again after uv add, an upgrade, or a large edit, or whenever these are not zero.

k.sync(embed=False) is a faster sync that stores new chunks without embeddings. Keyword search finds them straight away, and k.process_repo() and k.process_env() add the embeddings later.

Search before you write

env_context searches the installed packages. Searching them first often turns up a function that already does what you were about to write. Here a search for an atomic file write finds fsspec.utils.atomic_write and a Jupyter helper among others. Each hit is a dict with the chunk’s content and its metadata, which includes the dotted name (mod_name), the file path and the line number.

results = k.env_context('atomic write temp file permissions', limit=5)
for r in results:
    print(r['metadata']['mod_name'])
    print(' ', r['content'].splitlines()[0])
    print()

fsspec.implementations.webhdfs.WebHDFS._open
  def _open(

jupyter_server.services.contents.fileio.FileManagerMixin.atomic_writing
  def atomic_writing(self, os_path, *args, **kwargs):

setuptools._core_metadata.write_pkg_info
  def write_pkg_info(self, base_dir):

joblib._store_backends.StoreBackendMixin._concurrency_safe_write
  def _concurrency_safe_write(self, to_write, filename, write_func):

fsspec.utils.atomic_write
  def atomic_write(path: str, mode: str = "wb"):

Filters narrow a search. Write them in the query as key:value: package: for an installed package, path: and file: for a path glob, lang: for a language, and type: for the kind of definition, such as FunctionDef. Each filter excludes everything that does not match. A trailing !, as in package!:fastcore, means the same as without it.

A package name written as a plain word behaves differently in context, the combined search below. It only raises that package’s hits, and other packages still appear.

k.env_context('package:fastcore path:xtras atomic save', limit=8)   # only fastcore, only paths containing xtras
k.context('fastcore atomic save', limit=8)                          # fastcore ranked higher, others still listed

When a hit comes from a package you do not know, pkg_url gives the address of its repository or home page, which you can hand to a web search for its documentation or changelog:

from kosha.core import pkg_url
pkg_url('litesearch')

'https://github.com/Karthik777/litesearch'

Search the repo and the packages together

context is the search to reach for first. It searches your repo and the packages at once, merges the two lists, and reorders the hits with rules made for code: a query that names a symbol favours the chunk that defines it, and test files and re-exports rank lower. With graph=True each hit also carries its place in the call graph: pagerank, callers, callees and co_dispatched.

results = k.context('search code embeddings', limit=6, graph=True)
for r in results:
    m = r['metadata']
    print(f"{m['mod_name']}  L{m.get('lineno','?')}  "
          f"pr={r.get('pagerank') or 0:.4f}  callers={list(r.get('callers',[]))[:2]}")

chonkie.embeddings.auto.AutoEmbeddings  L13  pr=0.0000  callers=[]
kosha.core.process_repo  L283  pr=0.0000  callers=['kosha.core.Kosha']
kosha.graph._boost_embedded  L772  pr=0.0000  callers=['kosha.graph._apply_query_boost']
chonkie.handshakes.pinecone.PineconeHandshake.search  L201  pr=0.0000  callers=[]
transformers.models.squeezebert.modeling_squeezebert.SqueezeBertModel.set_input_embeddings  L434  pr=0.0000  callers=[]
chonkie.handshakes.elastic.ElasticHandshake.search  L152  pr=0.0000  callers=[]

PageRank, the algorithm Google first used to rank web pages, scores how much of the graph depends on a node. A function with a high score is called from many places, or from places that are themselves called a lot, so a change to it reaches further. Treat it as a warning to look before you edit. Most nodes score close to zero, as here.

Inspect what a change will touch

ni (node info) takes a full dotted name and returns the node’s graph record: its PageRank, the functions that call it, the functions it calls, and its co-dispatched peers. Call it before you change a function. In this example fastcore.basics.merge is called from two places in fastcore.script.

info = k.ni('fastcore.basics.merge')
print('pagerank:', info.get('pagerank', 0))
print('callers: ', list(info.get('callers', []))[:5])
print('callees: ', list(info.get('callees', []))[:5])
print('co_dispatched:', list(info.get('co_dispatched', []))[:5])

pagerank: None
callers:  ['fastcore.script.anno_parser', 'fastcore.script._run_cli']
callees:  []
co_dispatched: []

co_dispatched lists the node’s peers: functions registered together, in a route list, a handler table or a list of plugins. When you add a new member to such a group, the peers show the pattern to copy.

Choose where to make the change

where_to_add takes a description of new code and suggests where it should go. For each suggestion it gives the file, the line to insert after, and the node it would sit next to. When that node belongs to a co-dispatch group, the line is the end of the group’s last member in the file, so the new code lands beside its peers.

pts = k.where_to_add('add dynamic ast parsing for patched functions', limit=3)
for p in pts:
    co = ', '.join(p['co_dispatched'][:3])
    print(f"{p['path']}:{p['insert_after']}  ({p['node']})")
    if co: print(f'  peers: {co}')

/Users/71293/code/personal/orgs/kosha/kosha/graph.py:154  (kosha.graph.dyn_edges)
/Users/71293/code/personal/orgs/kosha/kosha/core.py:56  (kosha.core.parse)

Scan many results quickly

compact=True drops the source code from each hit and returns a slim dict with mod_name, path, lineno, docstring and sig, the first line of the definition. It is the cheap way to look over many results before reading a few in full. repo=False limits the search to packages.

hits = k.context('database search filter package:litesearch', limit=2,repo=False, compact=True)
for r in hits:
    sig = r.get('sig', '')
    doc = (r.get('docstring') or '')[:60]
    print(f"{r['mod_name']}  L{r.get('lineno','?')}")  
    if sig: print(f'  {sig}')
    if doc: print(f'  # {doc}')

litesearch.api.search  L100
  def search(self:Index,
  # Hybrid keyword + vector search over the chunk store.
litesearch.core.database  L397
  def database(pth_or_uri:str=':memory:',     # the database name or URL
  # Set up a database connection and load usearch extensions.

List a package’s public API

public_api lists the public names of a package or of one module in it. A name is public when the module lists it in __all__, or, for modules without __all__, when it has no leading underscore. Functions that fastcore’s @patch attaches to a class count too. Top-level names come before methods.

api = k.public_api('fastcore', limit=12)
for e in api:
    name = e.get('mod_name', '')
    doc = (e.get('docstring') or '')[:55]
    print(f"{name}" + (f'  # {doc}' if doc else ''))

fastcore.aio.CachedAwaitable  # Cache the result from an awaitable
fastcore.aio.acache  # Cache results of async function `f`
fastcore.aio.athreaded  # Run `f` in a worker thread, awaitably; use as `@athread
fastcore.aio.ctx_sync  # Use async context manager `acm` in a plain `with` block
fastcore.aio.disable_async_magics  # Undo `enable_async_magics` on `ip`
fastcore.aio.is_async_callable  # Check if `obj` is an async callable, handling `partial`
fastcore.aio.iter_sync  # Iterate async generator `agen` from sync code
fastcore.aio.mapa  # Async `map`; apply `f` (sync or async) to `items` (sync
fastcore.aio.maybe_aiter  # If `items` already async, return it; otherwise to_aiter
fastcore.aio.noopa  # Do nothing (async)
fastcore.aio.reawaitable  # Wraps the result of an asynchronous function into an ob
fastcore.aio.run_sync  # Run coroutine `coro` to completion from sync code and r

Trace a call path

Once you have a symbol or a package in hand, these queries walk the call graph. k.graphdb is the graph’s SQLite database, so you can query the graph_edges table directly, as the first cell does. short_path returns the shortest chain of calls from one node to another. api_call_paths finds call chains from the most used public functions of one package to those of another. dep_stack lists the packages a package imports, layer by layer. graph.ranked returns the nodes with the highest PageRank under a module prefix.

from fastcore.foundation import L

k.graphdb.t.graph_edges(where='callee like "%litesearch%"')[:2]

[{'caller': 'sanskrit.register_profiles',
  'callee': 'litesearch.data.register_profile',
  'kind': 'static',
  'confidence': 1.0},
 {'caller': 'sanskrit.register_profiles',
  'callee': 'litesearch.data.Profile',
  'kind': 'static',
  'confidence': 1.0}]

L(k.ni('kosha.core.env_context')['callees']).filter(lambda x: 'search' in x)

['litesearch.core.rerank_hits', 'litesearch.core.search']

# Shortest call chain between two graph nodes
k.short_path('kosha.core.env_context', 'litesearch.core.search')

['kosha.core.env_context', 'litesearch.core.search']

# Call paths from the top public functions of one package to those of another
paths = k.api_call_paths('kosha', 'litesearch', k=10)
for tgt, path in sorted(paths.items(), key=lambda x: len(x[1]))[:3]:
    print(f'{tgt}: {len(path)} hops')
    print('  ', ' → '.join(path))

# Layers of imported packages, breadth first from a seed package
k.dep_stack(seeds=['kosha'], depth=2)

[['kosha']]

# Top-k nodes by PageRank in a package
k.graph.ranked(k=5, module='fastcore')

[{'node': 'fastcore.all.L', 'pagerank': 0.00975}, {'node': 'fastcore.all.Path', 'pagerank': 0.00526}, {'node': 'fastcore.all.first', 'pagerank': 0.00254}, {'node': 'fastcore.all.ifnone', 'pagerank': 0.0013}, {'node': 'fastcore.all.patch', 'pagerank': 0.00129}]

Daemon mode

Each new Python process pays three to five seconds to load the embedding model before its first search. kosha daemon loads it once and then answers requests read from standard input, one JSON object per line, writing one JSON line per answer. It prints {"ok": true, "status": "ready"} when it is ready.

kosha daemon

Each request names a command and its arguments:

{"cmd": "context", "args": {"q": "embed a query", "limit": 10}}
{"ok": true, "result": [...]}

{"cmd": "short_path", "args": {"src": "kosha.core.Kosha.sync", "tgt": "litesearch.core.search"}}
{"ok": true, "result": [...]}

A failed request returns {"ok": false, "error": "..."}. The commands are sync, status, context, repo_context, env_context, ni, neighbors, short_path, top_nodes, public_api, api_call_paths, dep_stack, where_to_add and nuke, with the same arguments as the Kosha methods of those names. ni takes mod_name.

Watch mode

kosha watch watches the repository and re-indexes files as they change. It blocks until you press Ctrl-C. From Python, k.watch_repo() does the same.

kosha watch

Command line

Every query above is also a shell command. Output is markdown, and --as-json gives JSON for tools such as jq. Commands and flags use hyphens where the Python names use underscores, and kosha <command> --help lists a command’s flags.

kosha install                          # install SKILL.md into .agents/ and .claude/
kosha sync                             # index the repo, the packages and the call graph
kosha status                           # check whether the index is current
kosha context "embed a query" --as-json | jq '.[].metadata.mod_name'
kosha ni "fastcore.basics.merge"       # node info
kosha where-to-add "new route handler"
kosha public-api fastcore
kosha api-paths kosha litesearch
kosha daemon                           # long-running process for repeated queries

Agent skill

Kosha(install_skill=True), or kosha install from the shell, copies SKILL.md into .agents/skills/kosha/ and .claude/skills/kosha/. Commit .agents/skills/kosha/SKILL.md so every contributor’s agent picks it up.

pyskills

kosha also registers itself as a pyskill, kosha.skill, for agent hosts that load skills as Python modules.

MCP server

kosha-mcp serves the index over the Model Context Protocol (MCP), so Claude Code, Claude Desktop, Codex and other MCP clients can call it as a set of tools: status, sync, context, repo_context, env_context, where_to_add, node_info, neighbors, short_path, public_api, top_nodes, api_paths, dep_stack and pkg_url.

The server comes with the package. It indexes the repository it starts in and that repository’s environment, so it must start at the project root inside the project’s environment, and uv run takes care of both:

uv add --dev koshas

For Claude Code, run this inside the project:

claude mcp add kosha -- uv run kosha-mcp

For Codex, add this to ~/.codex/config.toml. Codex starts servers in your session’s working directory, so start Codex at the project root.

[mcp_servers.kosha]
command = "uv"
args = ["run", "kosha-mcp"]

Claude Desktop does not start in your project, so name the project in claude_desktop_config.json:

{"mcpServers": {"kosha": {"command": "uv", "args": ["run", "--project", "/path/to/your/repo", "kosha-mcp"]}}}

The server talks over standard input and output by default; kosha-mcp --http serves Streamable HTTP instead. The MCP page describes each tool.