Skip to main content
All examples assume you have initialized the SDK:
import os
from nia_py.sdk import NiaSDK

sdk = NiaSDK(api_key=os.environ["NIA_API_KEY"])

Repository Operations

Index a Repository

from nia_py import AuthenticatedClient
from nia_py.api.v2_api import index_repository_v2_v2_repositories_post
from nia_py.models import RepositoryRequest

client = AuthenticatedClient(
    base_url="https://apigcp.trynia.ai/v2",
    token=os.environ["NIA_API_KEY"],
)

result = index_repository_v2_v2_repositories_post.sync(
    client=client,
    body=RepositoryRequest(repository="vercel/ai", branch="main"),
)
print(f"Indexing started: {result}")
Or with the high-level SDK:
sdk.sources.create({"url": "https://github.com/vercel/ai", "branch": "main"})

Check Repository Status

from nia_py.api.v2_api import get_repository_status_v2_v2_repositories_repository_id_get

status = get_repository_status_v2_v2_repositories_repository_id_get.sync(
    client=client,
    repository_id="vercel/ai",
)
print(f"Status: {status}")

Browse Repository Tree

from nia_py.api.v2_api import get_repository_tree_v2_v2_repositories_repository_id_tree_get

tree = get_repository_tree_v2_v2_repositories_repository_id_tree_get.sync(
    client=client,
    repository_id="vercel/ai",
)
print(tree)

Read File Content

from nia_py.api.v2_api import get_repository_content_v2_v2_repositories_repository_id_content_get

content = get_repository_content_v2_v2_repositories_repository_id_content_get.sync(
    client=client,
    repository_id="vercel/ai",
    path="packages/ai/src/index.ts",
)
print(content)

Grep Search in Repository

from nia_py.api.v2_api import grep_repository_v2_v2_repositories_repository_id_grep_post
from nia_py.models import CodeGrepRequest

results = grep_repository_v2_v2_repositories_repository_id_grep_post.sync(
    client=client,
    repository_id="vercel/ai",
    body=CodeGrepRequest(
        pattern="async function",
        path="src/",
        context_lines=2,
        case_sensitive=False,
    ),
)
print(results)

result = sdk.search.query(
    messages=[{"role": "user", "content": "How does streaming work in the AI SDK?"}],
    repositories=["vercel/ai"],
    include_sources=True,
)
print(result)
Searches across all your indexed repositories and documentation automatically.
result = sdk.search.universal(
    query="How to implement tool calling?",
    top_k=10,
    include_repos=True,
    include_docs=True,
)
print(result)
result = sdk.search.web(query="latest LLM developments 2025")
print(result)

Deep Research

Multi-step research with AI analysis and citations.
result = sdk.search.deep(
    query="Compare React Server Components vs traditional SSR approaches",
    output_format="comparison table",
)
print(result)

Documentation Sources

Index Documentation

from nia_py.api.v2_api import create_data_source_v2_v2_data_sources_post
from nia_py.models import DataSourceRequest

result = create_data_source_v2_v2_data_sources_post.sync(
    client=client,
    body=DataSourceRequest(
        url="https://docs.anthropic.com",
        display_name="Anthropic Docs",
        crawl_entire_domain=True,
    ),
)
print(f"Indexing started: {result}")
Or with the high-level SDK:
sdk.sources.create({
    "url": "https://docs.anthropic.com",
    "display_name": "Anthropic Docs",
})

List Sources

# List all sources
all_sources = sdk.sources.list()

# Filter by type
repos = sdk.sources.list(type="repository")
docs = sdk.sources.list(type="documentation")

# Search by name
results = sdk.sources.list(query="anthropic")

Search Documentation

result = sdk.search.query(
    messages=[{"role": "user", "content": "How to use Claude's tool use feature?"}],
    data_sources=["Anthropic Docs"],
    search_mode="sources",
)
print(result)

Research Papers

Index an arXiv Paper

from nia_py.api.v2_api import index_research_paper_v2_v2_research_papers_post
from nia_py.models import ResearchPaperRequest

result = index_research_paper_v2_v2_research_papers_post.sync(
    client=client,
    body=ResearchPaperRequest(url="https://arxiv.org/abs/2303.08774"),
)
print(result)

Search source code of public packages on npm and PyPI without indexing.
from nia_py.api.v2_api import package_search_hybrid_v2_v2_package_search_hybrid_post
from nia_py.models import PackageSearchHybridRequest

result = package_search_hybrid_v2_v2_package_search_hybrid_post.sync(
    client=client,
    body=PackageSearchHybridRequest(
        registry="py_pi",
        package_name="openai",
        semantic_queries=["How does streaming work?"],
    ),
)
print(result)

Oracle Research

Create and Poll a Research Job

# Start a research job
job = sdk.oracle.create_job(
    query="Compare React Server Components vs traditional SSR",
    repositories=["vercel/next.js"],
    output_format="comparison table",
    model="claude-sonnet-4-5-20250929",
)
print(f"Job ID: {job['id']}, Status: {job['status']}")

# Poll until complete (default 10min timeout)
result = sdk.oracle.wait_for_job(
    job_id=job["id"],
    timeout_seconds=300.0,
    poll_interval_seconds=3.0,
)
print(result)

Stream Research Events

job = sdk.oracle.create_job(query="Analyze the caching strategy in Next.js App Router")

for event in sdk.oracle.stream_job_events(job_id=job["id"]):
    # Each event is a parsed JSON dict from the SSE stream
    event_type = event.get("type", "unknown")
    print(f"[{event_type}] {event}")

List and Manage Jobs

# List recent jobs
jobs = sdk.oracle.list_jobs(limit=10)

# Filter by status
completed = sdk.oracle.list_jobs(status="completed")

# Get a specific job
job = sdk.oracle.get_job(job_id="job-uuid")

Context Sharing

Save and retrieve context across sessions.
from nia_py.api.v2_api import (
    save_context_v2_v2_contexts_post,
    semantic_search_contexts_v2_v2_contexts_semantic_search_get,
)
from nia_py.models import ContextSaveRequest

# Save context
save_context_v2_v2_contexts_post.sync(
    client=client,
    body=ContextSaveRequest(
        title="Auth implementation notes",
        summary="Implemented OAuth2 flow with refresh tokens",
        content="Detailed implementation notes...",
        tags=["auth", "oauth"],
    ),
)

# Search contexts by semantic similarity
results = semantic_search_contexts_v2_v2_contexts_semantic_search_get.sync(
    client=client,
    query="authentication",
)
print(results)

GitHub Tracer

Live code search across GitHub repositories without indexing.
from nia_py.api.v2_api import create_tracer_job_v2_github_tracer_post
from nia_py.models import TracerRequest

job = create_tracer_job_v2_github_tracer_post.sync(
    client=client,
    body=TracerRequest(
        query="How does the streaming API work?",
        repositories=["vercel/ai"],
    ),
)
print(f"Tracer job: {job}")

Async Support (Python)

Every low-level API function has four variants:
from nia_py.api.v2_api import search_universal_v2_v2_search_universal_post
from nia_py.models import UniversalSearchRequest
from nia_py.types import Response

body = UniversalSearchRequest(query="test")

# Sync — returns parsed data or None
result = search_universal_v2_v2_search_universal_post.sync(client=client, body=body)

# Sync detailed — returns Response with status_code, headers, parsed
response: Response = search_universal_v2_v2_search_universal_post.sync_detailed(client=client, body=body)

# Async — returns parsed data or None
result = await search_universal_v2_v2_search_universal_post.asyncio(client=client, body=body)

# Async detailed — returns Response
response = await search_universal_v2_v2_search_universal_post.asyncio_detailed(client=client, body=body)