Python

Installation

The SDK is published to PyPI. Install it using pip or uv.

uv add larkup

pip install larkup

Initialization

Import the client and initialize it. The client will automatically fall back to LARKUP_API_URL and LARKUP_API_KEY environment variables if no arguments are passed. Both synchronous and asynchronous clients are available.

import os

from larkup import LarkupClient, LarkupClientOptions

options = LarkupClientOptions(
    base_url=os.getenv("LARKUP_API_URL"),
    api_key=os.getenv("LARKUP_API_KEY")
)

client = LarkupClient(options)

Basic Usage

Adding a Document

from larkup import Document

doc = Document(
    id="doc-1",
    text="Larkup is a flexible, high performance RAG pipeline.",
    title="Introduction"
)

response = client.add_document(doc)
print("Document added:", response["success"])

add_document() embeds and stores the document immediately. No second index call is required.

Bulk Indexing with Progress

documents = [
    Document(title="Guide", text="Deployment guide content"),
    Document(title="FAQ", text="Frequently asked questions"),
]

for progress in client.index_documents(
    documents,
    mode="parallel",
    concurrency=4,
    continue_on_error=True,
):
    print(progress.percent, progress.succeeded, progress.failed)

Use mode="sequential" for strict request ordering or embedding providers with strict rate limits. Parallel mode controls client request scheduling only. The server retains its configured index type, vector store, embedding model, and chunking strategy.

Querying

results = client.query("What is Larkup?", top_k=5)

for hit in results.hits:
    print(f"Score: {hit.score} | Text: {hit.text}")

Streaming Chat

for event in client.chat("Summarize the deployment guide."):
    if event.type == "text-delta":
        print(event.text or "", end="", flush=True)

answer = client.chat_text("What does the API key protect?")

The asynchronous client provides matching chat() and chat_text() methods:

from larkup import AsyncLarkupClient

async with AsyncLarkupClient(options) as client:
    async for event in client.chat("What is Larkup?"):
        if event.type == "text-delta":
            print(event.text or "", end="")

The asynchronous client also exposes index_documents() as an async iterator.

Agent Server

Use LarkupAgentClient when the Larkup Server toggle is set to Agent. It lists loaded tools and consumes the Agent’s AI SDK UI-message stream.

import os

from larkup import LarkupAgentClient

agent = LarkupAgentClient(
    base_url="http://localhost:8080",
    api_key=os.environ["LARKUP_API_KEY"],
)
for capability in agent.capabilities():
    print(capability.name, capability.tools)
print(agent.configuration().systemPrompt)
print(agent.sandbox())
for text in agent.stream_text("What can you do?"):
    print(text, end="", flush=True)

Use capabilities() to inspect grouped integrations—selected built-ins, skills, sandbox, one entry per MCP connection, and plugins. A capability marked configured is selected in the Project but is not executable by this runtime; active entries are loaded. tools() remains available when you need the raw tool list. configuration() returns the UI prompt, selected tools, skills, and sandbox configuration; sandbox() checks readiness without exposing credentials. chat_text() collects through the streaming path, so slow tool calls do not hit the buffered-response timeout. Agent clients use a 120-second timeout by default; pass timeout= to adjust it. chat() exposes the raw HTTP response.

Choose a chat model

Every generated runtime exposes GET /models. Inspect its available choices through the SDK, then use modelId for one request without changing the deployment default.

from larkup import AgentChatRequest

catalog = agent.chat_model_catalog()
print(agent.chat_providers())

models = agent.chat_models("anthropic")
if models:
    answer = agent.chat_text(
        AgentChatRequest(
            messages=[{"role": "user", "content": "Summarize the project."}],
            # Optional. Omit it to use catalog.configuredProvider.
            provider=catalog.configuredProvider,
            modelId=models[0].id,
        )
    )
    print(answer)

An AI Gateway runtime can select any language model listed by its catalog. A runtime configured with a direct provider accepts only that provider’s model IDs, so its configured API key is never used with another vendor. Browse the current catalog and provider capabilities in the Vercel AI Gateway model catalog.

Corpus Inspection and Export

from larkup import CorpusFilter, CorpusRequest

summary = client.corpus_summary()
corpus = client.corpus(
    CorpusRequest(
        filter=CorpusFilter(titleContains="deployment"),
        limit=50,
        includeContent=True,
    )
)
jsonl = client.export_corpus("jsonl")

Media

media_list = client.list_media()
media = client.get_media("media-1")
job_status = client.get_media_job_status("job-1")
client.delete_media("media-1")

Marketplace Hub

from larkup import LarkupHubClient

with LarkupHubClient() as hub:
    tools = hub.list_tools(category="media", search="video")
    tool = hub.get_tool("video-audio")

The matching AsyncLarkupHubClient supports asynchronous applications. Tool installation changes the Larkup host and therefore remains a CLI operation:

larkup marketplace install video-audio

API Coverage

SDK method	RAG server operation
`health()`	`GET /health`
`open_api()`	`GET /openapi.json`
`query()`	`POST /query`
`chat()` / `chat_text()`	`POST /chat`
`list_documents()` / `get_document()`	`GET /documents`
`add_document()` / `index_documents()`	`POST /documents`
`update_document()` / `delete_document()`	`PUT` / `DELETE /documents/:id`
`scrape()`	`POST /scrape`
`corpus_summary()` / `corpus()`	`GET /corpus/summary`, `POST /corpus`
`export_corpus()`	`POST /corpus/export`
`list_media()` / `get_media()` / `delete_media()`	`GET /media`, `GET /media/:id`, `DELETE /media/:id`
`get_media_job_status()`	`GET /media/jobs/:id`
`LarkupAgentClient.tools()`	`GET /agent/tools`
`LarkupAgentClient.capabilities()`	`GET /agent/capabilities`
`LarkupAgentClient.configuration()`	`GET /agent/configuration` (admin)
`LarkupAgentClient.sandbox()`	`GET /agent/sandbox` (admin)
`LarkupAgentClient.chat()` / `stream_text()` / `chat_text()`	`POST /chat` (Agent profile)

Local Demo

The repository includes a complete demo configured for http://localhost:8080:

cd apps/sdk/py-sdk
uv run python examples/rag_server.py

Integrations

For connecting to LangChain Python, see the dedicated LangChain Integration guide.

Overview

Official

Integrations

Installation

Initialization

Basic Usage

Adding a Document

Bulk Indexing with Progress

Querying

Streaming Chat

Agent Server

Choose a chat model

Corpus Inspection and Export

Media

Marketplace Hub

API Coverage

Local Demo

Integrations

​Installation

​Initialization

​Basic Usage

​Adding a Document

​Bulk Indexing with Progress

​Querying

​Streaming Chat

​Agent Server

​Choose a chat model

​Corpus Inspection and Export

​Media

​Marketplace Hub

​API Coverage

​Local Demo

​Integrations

Installation

Initialization

Basic Usage

Adding a Document

Bulk Indexing with Progress

Querying

Streaming Chat

Agent Server

Choose a chat model

Corpus Inspection and Export

Media

Marketplace Hub

API Coverage

Local Demo

Integrations