kosha

kosha

Find the code you need before you write it.

Kosha keeps a searchable memory of your repository and installed packages. Start with semantic search; add call-graph context when you need to understand the impact of a change. It works locally, uses no LLM, and returns code you can inspect.

Install

kosha is a dev dependency. It indexes at development time so AI coding assistants can search it.

uv add --dev kosha

One-time project setup, which installs SKILL.md so every agent picks up the skill automatically:

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

Start a session

Create one index for the repository and the packages it uses. Later syncs compare source fingerprints and skip unchanged files.

k = Kosha()
k.sync()

k.sync(graph=False) skips the call graph. graph_mode='full' extracts graph batches in worker processes. graph_metrics=False defers PageRank and degree updates; call k.graph.recompute_metrics() after the graph updates finish.

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}

Re-run k.sync() after uv add, version bumps, or significant code changes. If stale_files > 0 or stale_pkgs is non-empty, sync before querying.

Use k.sync(embed=False) to rebuild the call graph on an existing DB without re-embedding, useful after a kosha update that changes graph logic.

Search before you write

Search installed packages first. It often finds an existing function or pattern before you add another one.

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"):

Package names in the query (package:fastcore, or a bare package word) are soft-boosted, matching results rank higher but other packages still appear. Use package!:fastcore to hard-filter to a single package. path:, lang:, type: tokens are hard filters that narrow further:

k.env_context('package:fastcore path:xtras atomic save', limit=8)   # boost fastcore, keep others
k.context('atomic save package!:fastcore', limit=8)                 # fastcore only

Need more info on a package? Call pkg_url to get its repo/docs URL, then use websearch for changelogs, API docs, or migration guides:

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

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

Find a local pattern

Use repository and package search together when the task changes existing behaviour.

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 = blast radius, higher means more things depend on it, touch carefully.

Inspect the impact

Node information shows callers, callees, peers, and PageRank before you change a symbol.

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 sibling functions registered together (route groups, handler tables, plugin lists), the pattern to follow when adding a new one.

Choose where to make the change

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)

Triage, scan many results quickly

compact=True strips full code bodies and returns slim dicts for fast scanning.

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.

Public API surface

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

Use these graph queries after you have a symbol or package in hand. They show a shortest call chain, public API paths, dependency layers, and the most connected nodes.

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']

# Public-API → public-API call paths between two packages
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))

# BFS dependency layers from a seed package, ordered by coupling strength
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, warm kernel for sessions

The first kosha call in a process pays a 3–5s embedder cold-start. kosha daemon keeps a warm process running and routes JSON requests over stdin/stdout, so subsequent calls are immediate.

kosha daemon &     # start once per session

Then send newline-delimited JSON requests:

→ {"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":[…]}

Available commands: sync, status, context, repo_context, env_context, ni, neighbors, short_path, top_nodes, public_api, api_call_paths, dep_stack, where_to_add.

Live watch mode

Re-index the repo incrementally on every file change (blocking, Ctrl-C to stop):

kosha watch

or programmatically:

k.watch_repo()

CLI

Shell access to everything. Markdown by default; --as_json pipes into jq.

kosha install                         # install SKILL.md to .agents/ and .claude/
kosha sync  # index repo + env + call graph
kosha status # check index freshness
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 # persistent kernel, warm for all session calls

Harness install

Kosha(install_skill=True)   # installs to .agents/ and .claude/

Commit .agents/skills/kosha/SKILL.md so every contributor picks up the skill automatically.

pyskills

kosha registers as a pyskill (kosha.skill) for Python-native LLM hosts.

MCP server

kosha-mcp exposes the index over the Model Context Protocol, so Claude Code, Claude Desktop, Codex, and any other MCP client can query it directly, status/sync, context/repo_context/env_context, node_info/short_path/api_paths, where_to_add, and more.

The MCP server ships with kosha (no extra needed). kosha indexes the current repo and its venv, so the server must launch from the project root with the project’s environment, uv run does both:

uv add --dev koshas

Claude Code (run inside the project)

claude mcp add kosha -- uv run kosha-mcp

Codex (~/.codex/config.toml; Codex launches servers from your session’s working directory, so start it at the project root)

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

Claude Desktop (claude_desktop_config.json, pin the project explicitly)

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

The server speaks stdio by default (kosha-mcp --http for Streamable HTTP). See the mcp docs for the full tool list.