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 — 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) is the shortest refresh when you only need semantic search. graph_mode='full' asks for pyan3’s broader static analysis. Pass graph_metrics=False when you want to defer PageRank for a large graph refresh.

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, 1519.12it/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, 100.77pkg/s]
Updating packages:  50%|██████████████████████████████████████████████████                                                  | 1/2 [00:00<00:00,  2.57pkg/s]Updating packages: 100%|████████████████████████████████████████████████████████████████████████████████████████████████████| 2/2 [00:00<00:00,  4.46pkg/s]Updating packages: 100%|████████████████████████████████████████████████████████████████████████████████████████████████████| 2/2 [00:00<00:00,  4.01pkg/s]

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

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

k.status()

{'files': 4,
 'packages': 414,
 'graph_nodes': 10425,
 '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',0):.4f}  callers={list(r.get('callers',[]))[:2]}")

chonkie.embeddings.auto.AutoEmbeddings  L13  pr=0.0000  callers=[]
kosha.core.process_repo  L283  pr=0.0001  callers=['kosha.core.Kosha']
chonkie.handshakes.pinecone.PineconeHandshake.search  L201  pr=0.0000  callers=[]
transformers.models.squeezebert.modeling_squeezebert.SqueezeBertModel.set_input_embeddings  L434  pr=0.0000  callers=[]
kosha.graph._boost_embedded  L772  pr=0.0001  callers=['kosha.graph._apply_query_boost']
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: 0
callers:  []
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:1034  (kosha.graph._fast_edges)
/Users/71293/code/personal/orgs/kosha/kosha/core.py:56  (kosha.core.parse)
/Users/71293/code/personal/orgs/kosha/kosha/graph.py:154  (kosha.graph.dyn_edges)

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.run_sync  # Run coroutine `coro` to completion from sync code and r
fastcore.aio.iter_sync  # Iterate async generator `agen` from sync code
fastcore.aio.ctx_sync  # Use async context manager `acm` in a plain `with` block
fastcore.aio.athreaded  # Run `f` in a worker thread, awaitably; use as `@athread
fastcore.aio.then  # Pipe `x` through each of `fs`, awaiting values as neede
fastcore.aio.acache  # Cache results of async function `f`
fastcore.aio.CachedAwaitable  # Cache the result from an awaitable
fastcore.aio.reawaitable  # Wraps the result of an asynchronous function into an ob
fastcore.aio.is_async_callable  # Check if `obj` is an async callable, handling `partial`
fastcore.aio.to_aiter  # Async yield each item in `items` with `asyncio.sleep(0)
fastcore.aio.maybe_aiter  # If `items` already async, return it; otherwise to_aiter
fastcore.aio.mapa  # Async `map`; apply `f` (sync or async) to `items` (sync

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': 'core.database',
  'callee': 'litesearch.sanskrit.register_sanskrit',
  'kind': 'static',
  'confidence': 1.0},
 {'caller': 'sanskrit.register_profiles',
  'callee': 'litesearch.data.register_profile',
  'kind': 'static',
  'confidence': 1.0}]

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

[]

# Shortest call chain between two graph nodes
k.short_path('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.00257}, {'node': 'fastcore.all.Path', 'pagerank': 0.00116}, {'node': 'fastcore.all.first', 'pagerank': 0.00056}, {'node': 'fastcore.all.ifnone', 'pagerank': 0.00039}, {'node': 'fastcore.all.patch', 'pagerank': 0.00037}]

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.