Python

The grepvine package covers the whole API with two clients: Client, which blocks, and AsyncClient, which has the same methods as coroutines. Indexing and searching show each call in every client. This page covers what is specific to Python.

Install.

pip install grepvine

The package needs Python 3.11 or later and ships type hints. With uv: uv add grepvine.

Create a client.

Sync

import os
import grepvine

with grepvine.Client(api_key=os.environ["GREPVINE_API_KEY"]) as client:
    index = client.indexes.get("acme", "support")
    print(index.document_count)

Async

import asyncio
import os
import grepvine

async def main() -> None:
    async with grepvine.AsyncClient(api_key=os.environ["GREPVINE_API_KEY"]) as client:
        index = await client.indexes.get("acme", "support")
        print(index.document_count)

asyncio.run(main())

Both clients take the same keyword arguments:

Argument	Default	Meaning
api_key	None	Sent as a bearer token. Without one, only device login and account sign-in work.
endpoint	https://api.grepvine.ai	The service URL, with or without /v1.
timeout	60	Seconds per request, or an httpx.Timeout.
retry	RetryPolicy()	How failed requests are retried. See retries.
transport	httpx’s default	An httpx transport, such as httpx.MockTransport in tests.

• Pass configuration in. The package reads no environment variables or configuration files.
• Close the client. Use it as a context manager, or call close() on a Client and aclose() on an AsyncClient.
• Reuse one client. A Client can be shared between threads. Create an AsyncClient inside the event loop that uses it. An iterator from paginate isn’t thread-safe.

Make calls.

The client’s resources follow the API: organizations (with organizations.members), indexes, documents, search, api_keys, account and device_login. Each method takes the path parameters organization, index and key by position, and everything else by name. A request body is a model from grepvine.models, passed as body=.

from grepvine.models import QueryRequest, QueryRequestMode

query = QueryRequest(pattern="connection reset by peer", mode=QueryRequestMode.LITERAL, case_sensitive=False)
page = client.search.query("acme", "support", body=query, amount=100)
for hit in page.results:
    print(hit.key, hit.line_number)

• Two names end in _. They would shadow builtins: filter_ on QueryRequest and range_ on documents.get.
• Absent fields are UNSET. An optional response field the service left out is UNSET, which is falsy: for line in hit.lines_after or [].
• Methods return the response model, or None when the response has no body.

Page through results.

Each list method and search.query returns one page. grepvine.paginate takes the method and its arguments by name and yields every result on every page. With an AsyncClient method, iterate with async for.

for document in grepvine.paginate(client.documents.list, organization="acme", index="support", prefix="tickets/2026/10/"):
    print(document.key, document.content_bytes)

To page by hand, pass page=page.pagination.next_page_id with the same arguments, including amount, until page.pagination.has_next is false. A page token lasts 15 minutes. An expired or mismatched token raises InvalidPageError.

Handle errors.

Every exception the package raises is a GrepvineError. An error answer from the service raises an APIError subclass with status_code, code, message and request_id. code is the API’s error code, such as invalid_filter. str(error) includes the code and the request ID.

Exception	Status	Raised when
InvalidRequestError	400	The input was rejected, such as a pattern or filter that doesn’t parse. Its subclass InvalidPageError is for page tokens.
AuthenticationError	401	The API key is missing, wrong or revoked.
PermissionDeniedError	403	The key’s role doesn’t allow the action.
NotFoundError	404	The organization, index or document doesn’t exist.
ConflictError	409	The request conflicts with the current state, such as creating an index that exists.
PreconditionFailedError	412	A condition failed on documents.put or documents.delete. Batch writes report it per document instead.
PayloadTooLargeError	413	A document or batch is over its size limit.
RangeNotSatisfiableError	416	A byte range doesn’t overlap the document.
RateLimitedError	429	The client waited out Retry-After for longer than its retry policy allows.
ServerError	5xx	The service failed on every attempt.

Three errors don’t come from a response. TransportError means the request couldn’t be delivered after every attempt, ConfigurationError means the endpoint isn’t an http or https URL, and PartialBatchError comes from put_many and is described in indexing.

This creates an index only if it doesn’t exist yet:

from grepvine.models import CreateIndexRequest

try:
    client.indexes.create("acme", body=CreateIndexRequest(name="support"))
except grepvine.ConflictError:
    pass
client.indexes.wait_until_ready("acme", "support")

Retries.

The client retries on its own:

• 429 waits for the Retry-After interval, varied by up to 25% either way, at most 60 seconds at a time and 600 seconds in total per request. After that it raises RateLimitedError.
• 5xx and connection failures are retried for up to 3 attempts in total, after 0.1 and then 0.2 seconds.
• Writes carry an Idempotency-Key that stays the same across attempts, so a retried write is applied once.

An agent answering a person shouldn’t wait ten minutes for a search. Shorten the policy for interactive use:

client = grepvine.Client(
    api_key=os.environ["GREPVINE_API_KEY"],
    timeout=10,
    retry=grepvine.RetryPolicy(max_attempts=5, max_throttled_wait=30),
)

RetryPolicy also takes backoff, default_retry_after, max_retry_after and retry_after_jitter.

Common patterns.

Give an agent a search tool.

The function below is the body of a search tool. It takes a pattern and an optional path prefix, caps the number of matches, and returns grep-style text with two lines of context. A rejected pattern or filter returns the service’s message, such as filter byte 9: expected a value, so the model can correct its query on the next turn. Register it with your agent framework as a tool. In the async version, client is an AsyncClient created in the agent’s event loop.

def format_hit(hit) -> str:
    before = [f"{hit.key}-{line.line_number}- {line.text}" for line in hit.lines_before or []]
    after = [f"{hit.key}-{line.line_number}- {line.text}" for line in hit.lines_after or []]
    return "\n".join([*before, f"{hit.key}:{hit.line_number}: {hit.text}", *after])

Sync

from itertools import islice
from grepvine.models import QueryRequest

def search_code(pattern: str, path_prefix: str = "", max_results: int = 50) -> str:
    """Search acme/monorepo with an RE2 pattern. Returns matches with two lines of context."""
    query = QueryRequest(pattern=pattern, lines_before=2, lines_after=2)
    if path_prefix:
        quoted = path_prefix.replace("'", "''")
        query.filter_ = f"path STARTSWITH '{quoted}'"
    hits = grepvine.paginate(client.search.query, organization="acme", index="monorepo", body=query)
    try:
        found = [format_hit(hit) for hit in islice(hits, max_results)]
    except grepvine.InvalidRequestError as error:
        return f"The search was rejected: {error.message}"
    return "\n--\n".join(found) or "No matches."

Async

from grepvine.models import QueryRequest

async def search_code(pattern: str, path_prefix: str = "", max_results: int = 50) -> str:
    """Search acme/monorepo with an RE2 pattern. Returns matches with two lines of context."""
    query = QueryRequest(pattern=pattern, lines_before=2, lines_after=2)
    if path_prefix:
        quoted = path_prefix.replace("'", "''")
        query.filter_ = f"path STARTSWITH '{quoted}'"
    found: list[str] = []
    try:
        async for hit in grepvine.paginate(
            client.search.query, organization="acme", index="monorepo", body=query
        ):
            found.append(format_hit(hit))
            if len(found) == max_results:
                break
    except grepvine.InvalidRequestError as error:
        return f"The search was rejected: {error.message}"
    return "\n--\n".join(found) or "No matches."

Doubling single quotes keeps a prefix such as docs/o'brien/ from breaking the filter. Best practices explains the cap.

Write batches in parallel.

put_many sends its requests one after another, and it builds every request body before it sends the first one. For a large import, split the documents into slices and send several slices at once:

from datetime import datetime, UTC
from pathlib import Path

root = Path("exports/tickets")
documents = [
    {
        "key": path.relative_to(root).as_posix(),
        "content": path.read_text(encoding="utf-8"),
        "metadata": {"mtime": datetime.fromtimestamp(path.stat().st_mtime, UTC)},
    }
    for path in sorted(root.rglob("*.md"))
]
slices = [documents[i : i + 2048] for i in range(0, len(documents), 2048)]

A suffix per slice keeps each idempotency key stable when you rerun the import.

Sync

from concurrent.futures import ThreadPoolExecutor

def send(n, part):
    return client.documents.put_many(
        "acme", "support", part, idempotency_key=f"support-import-2026-10-14-{n}"
    )

with ThreadPoolExecutor(max_workers=8) as pool:
    results = [result for part in pool.map(send, range(len(slices)), slices) for result in part]

Async

import asyncio

async def upload(slices) -> list:
    limit = asyncio.Semaphore(8)
    async with grepvine.AsyncClient(api_key=os.environ["GREPVINE_API_KEY"]) as client:

        async def send(n, part):
            async with limit:
                return await client.documents.put_many(
                    "acme", "support", part, idempotency_key=f"support-import-2026-10-14-{n}"
                )

        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(send(n, part)) for n, part in enumerate(slices)]
    return [result for task in tasks for result in task.result()]

results = asyncio.run(upload(slices))

Check each result’s error as in indexing. For a single directory, gv upload does the same with 16 batches in flight.

Search for several patterns at once.

With AsyncClient, independent searches run concurrently. This counts possible leaked credentials of three kinds:

patterns = [r"AKIA[0-9A-Z]{16}", r"-----BEGIN [A-Z ]*PRIVATE KEY-----", r"xox[bp]-[0-9A-Za-z-]{10,}"]

async def count(client: grepvine.AsyncClient, pattern: str) -> int:
    query = QueryRequest(pattern=pattern)
    return sum([1 async for _ in grepvine.paginate(
        client.search.query, organization="acme", index="monorepo", body=query
    )])

async def scan() -> None:
    async with (
        grepvine.AsyncClient(api_key=os.environ["GREPVINE_API_KEY"]) as client,
        asyncio.TaskGroup() as group,
    ):
        tasks = {pattern: group.create_task(count(client, pattern)) for pattern in patterns}
    for pattern, task in tasks.items():
        print(f"{task.result():5}  {pattern}")

asyncio.run(scan())

Read the matched text.

Offsets in a result count bytes, and Python strings index by code point. Encode text before you slice it, or a match after a non-ASCII character comes out shifted:

def matched_text(hit) -> list[str]:
    data = hit.text.encode()
    return [
        data[m.offset - hit.text_start_offset :][: m.length].decode(errors="replace")
        for m in hit.matches
    ]

documents.get with range_ returns part of a document, as searching shows. documents.get reads the whole response into memory. To save a large document to a file, stream it with client.http, the httpx.Client the client sends its requests with. It already has the endpoint, credentials, timeout and retries, and paths are relative to /v1:

from urllib.parse import quote

path = f"/organizations/acme/indexes/support/objects/{quote('exports/2026-10.csv', safe='')}"
with client.http.stream("GET", path) as response, open("2026-10.csv", "wb") as file:
    for chunk in response.iter_bytes():
        file.write(chunk)

Test without the service.

Pass an httpx.MockTransport as transport to answer requests in-process:

import httpx
import grepvine
from grepvine.models import QueryRequest

def test_search_without_matches():
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/v1/organizations/acme/indexes/monorepo/query"
        return httpx.Response(200, json={"results": [], "pagination": {"next_page_id": "", "has_next": False}})

    client = grepvine.Client(api_key="gvk_test", transport=httpx.MockTransport(handler))
    page = client.search.query("acme", "monorepo", body=QueryRequest(pattern="OrderNotFound"))
    assert page.results == []

Error answers from the handler raise the same exceptions as real ones. A 429 or 5xx answer is retried, so a test of either should pass retry=grepvine.RetryPolicy(max_attempts=1, max_throttled_wait=0) to fail at once.

API reference.

The full module reference is generated from the package: grepvine-ai.github.io/grepvine-python-sdk.