Python SDK

The OpenViking Python SDK provides both synchronous and asynchronous clients for embedded and HTTP modes.

Installation

pip install openviking

Client Types

Client	Mode	Use Case
`SyncOpenViking`	Embedded	Single-process apps, scripts
`AsyncOpenViking`	Embedded	Async apps with local storage
`SyncHTTPClient`	HTTP	Connect to remote server (sync)
`AsyncHTTPClient`	HTTP	Connect to remote server (async)

OpenViking is an alias for SyncOpenViking for convenience.

Quick Start

Embedded Mode
HTTP Mode
Async

import openviking as ov

# Initialize client
client = ov.OpenViking(path="./data")
client.initialize()

# Add resource
result = client.add_resource(
    path="https://example.com/docs",
    wait=True
)

# Search
results = client.find("how to use openviking")
for r in results.resources:
    print(f"{r.uri}: {r.score}")

client.close()

import openviking as ov

# Connect to server
client = ov.SyncHTTPClient(
    url="http://localhost:1933",
    api_key="your-key",
    agent_id="my-agent"
)

client.initialize()

# Add resource
result = client.add_resource(
    path="https://example.com/docs",
    wait=True
)

# Search
results = client.find("how to use openviking")

client.close()

import openviking as ov

async def main():
    client = ov.AsyncOpenViking(path="./data")
    await client.initialize()
    
    result = await client.add_resource(
        path="https://example.com/docs",
        wait=True
    )
    
    results = await client.find("how to use openviking")
    
    await client.close()

import asyncio
asyncio.run(main())

Client Initialization

SyncOpenViking / AsyncOpenViking

client = ov.SyncOpenViking(path: Optional[str] = None)

Parameters:

Parameter	Type	Description	Default
`path`	`str`	Local storage path (overrides `ov.conf`)	`None`

SyncHTTPClient / AsyncHTTPClient

client = ov.SyncHTTPClient(
    url: Optional[str] = None,
    api_key: Optional[str] = None,
    agent_id: Optional[str] = None,
    timeout: float = 60.0
)

Parameters:

Parameter	Type	Description	Default
`url`	`str`	Server URL (auto-loads from `ovcli.conf` if `None`)	`None`
`api_key`	`str`	API key (auto-loads from `ovcli.conf` if `None`)	`None`
`agent_id`	`str`	Agent identifier for isolation	`None`
`timeout`	`float`	HTTP request timeout in seconds	`60.0`

Core Methods

Lifecycle

initialize()

client.initialize() -> None

Initialize storage and indexes. Must be called before other operations.Example:

client = ov.OpenViking(path="./data")
client.initialize()

close()

client.close() -> None

Close client and release resources.Example:

try:
    client.initialize()
    # ... operations ...
finally:
    client.close()

reset()

SyncOpenViking.reset() -> None  # Class method

Reset singleton instance (mainly for testing).Example:

SyncOpenViking.reset()
client = ov.OpenViking(path="./new_data")

Resource Management

add_resource()

client.add_resource(
    path: str,
    target: Optional[str] = None,
    reason: str = "",
    instruction: str = "",
    wait: bool = False,
    timeout: float = None,
    build_index: bool = True,
    summarize: bool = False,
    **kwargs
) -> Dict[str, Any]

Add a resource (file, URL, or directory) to OpenViking.Parameters:

Parameter	Type	Description
`path`	`str`	Local file path or URL
`target`	`str`	Target path in VikingFS (e.g., `"kb/docs"`)
`reason`	`str`	Context/reason for adding
`instruction`	`str`	Processing instruction
`wait`	`bool`	Wait for processing to complete
`timeout`	`float`	Wait timeout in seconds
`build_index`	`bool`	Build vector index immediately
`summarize`	`bool`	Generate summary

Returns:

{
    "root_uri": "viking://resources/example",
    "status": "processing"
}

Example:

# Add URL and wait
result = client.add_resource(
    path="https://github.com/volcengine/OpenViking/README.md",
    reason="Project documentation",
    wait=True,
    timeout=120
)
root_uri = result["root_uri"]

# Add local directory
result = client.add_resource(
    path="./docs",
    target="kb/project-docs",
    wait=True
)

add_skill()

client.add_skill(
    data: Any,
    wait: bool = False,
    timeout: float = None
) -> Dict[str, Any]

Add a skill to OpenViking.Example:

skill_data = {
    "name": "example-skill",
    "description": "Example skill"
}
result = client.add_skill(skill_data, wait=True)

rm()

client.rm(
    uri: str,
    recursive: bool = False
) -> None

Delete a resource.Example:

# Delete file
client.rm("viking://resources/example.md")

# Delete directory recursively
client.rm("viking://resources/docs", recursive=True)

mv()

client.mv(
    from_uri: str,
    to_uri: str
) -> None

Move or rename a resource.Example:

client.mv(
    "viking://resources/old.md",
    "viking://resources/new.md"
)

wait_processed()

client.wait_processed(
    timeout: float = None
) -> Dict[str, Any]

Wait for all queued processing to complete.Example:

client.add_resource("https://example.com/large-doc")
client.wait_processed(timeout=300)  # Wait up to 5 minutes

Search and Retrieval

find()

client.find(
    query: str,
    target_uri: str = "",
    limit: int = 10,
    score_threshold: Optional[float] = None,
    filter: Optional[Dict] = None
) -> FindResult

Semantic search (quick retrieval without session context).Parameters:

Parameter	Type	Description
`query`	`str`	Search query
`target_uri`	`str`	Scope to directory
`limit`	`int`	Max results
`score_threshold`	`float`	Minimum similarity score
`filter`	`dict`	Metadata filters

Returns: FindResult with:

resources: List of SearchResult objects
Each SearchResult has: uri, score, content

Example:

# Basic search
results = client.find("how to deploy openviking")
for r in results.resources:
    print(f"{r.uri}: {r.score:.4f}")

# Search with filters
results = client.find(
    query="authentication guide",
    target_uri="viking://resources/docs",
    limit=5,
    score_threshold=0.7
)

search()

client.search(
    query: str,
    target_uri: str = "",
    session: Optional[Session] = None,
    session_id: Optional[str] = None,
    limit: int = 10,
    score_threshold: Optional[float] = None,
    filter: Optional[Dict] = None
) -> FindResult

Context-aware search with session (intent analysis + hierarchical retrieval).Example:

session = client.session()
run_async(session.add_message(role="user", content="Tell me about OpenViking"))

# Search with session context
results = client.search(
    query="how to use it",
    session=session,
    limit=5
)

grep()

client.grep(
    uri: str,
    pattern: str,
    case_insensitive: bool = False
) -> Dict

Content pattern search (regex).Example:

# Case-sensitive search
results = client.grep(
    uri="viking://resources",
    pattern="OpenViking"
)

# Case-insensitive
results = client.grep(
    uri="viking://resources",
    pattern="openviking",
    case_insensitive=True
)

glob()

client.glob(
    pattern: str,
    uri: str = "viking://"
) -> Dict

File pattern matching.Example:

# Find all markdown files
results = client.glob("**/*.md", uri="viking://resources")
for match in results["matches"]:
    print(match)

# Find specific pattern
results = client.glob("**/config*.json")

File System Operations

ls()

client.ls(
    uri: str,
    simple: bool = False,
    recursive: bool = False
) -> List[Any]

List directory contents.Example:

# Basic listing
entries = client.ls("viking://resources")
for entry in entries:
    print(f"{entry['name']} - {entry['isDir']}")

# Simple mode (paths only)
paths = client.ls("viking://resources", simple=True)

# Recursive
all_entries = client.ls("viking://resources", recursive=True)

tree()

client.tree(
    uri: str,
    **kwargs
) -> Dict

Get directory tree structure.Example:

tree = client.tree("viking://resources")
print(f"Nodes: {len(tree.get('children', []))}")

mkdir()

client.mkdir(uri: str) -> None

Create a directory.Example:

client.mkdir("viking://resources/new-folder")

stat()

client.stat(uri: str) -> Dict

Get resource metadata.Example:

info = client.stat("viking://resources/example.md")
print(f"Size: {info['size']}")
print(f"Modified: {info['modified']}")

Content Access

read()

client.read(
    uri: str,
    offset: int = 0,
    limit: int = -1
) -> str

Read L2 (full content).Example:

# Read entire file
content = client.read("viking://resources/example.md")

# Read with pagination
content = client.read("viking://resources/large.txt", offset=0, limit=1000)

abstract()

client.abstract(uri: str) -> str

Read L0 abstract (~100 token summary).Example:

abstract = client.abstract("viking://resources/docs")
print(f"Summary: {abstract}")

overview()

client.overview(uri: str) -> str

Read L1 overview (~2k token overview with navigation).Example:

overview = client.overview("viking://resources/docs")
print(overview)

Session Management

session()

client.session(
    session_id: Optional[str] = None,
    must_exist: bool = False
) -> Session

Create new session or load existing.Example:

# Create new session
session = client.session()

# Load existing
session = client.session(session_id="abc123", must_exist=True)

session_exists()

client.session_exists(session_id: str) -> bool

Check if session exists.Example:

if client.session_exists("abc123"):
    session = client.session("abc123")

create_session()

client.create_session() -> Dict[str, Any]

Create new session (returns dict).Example:

result = client.create_session()
session_id = result["session_id"]

list_sessions()

client.list_sessions() -> List[Any]

List all sessions.Example:

sessions = client.list_sessions()
for s in sessions:
    print(f"{s['session_id']}: {s['created_at']}")

get_session()

client.get_session(session_id: str) -> Dict[str, Any]

Get session details.Example:

details = client.get_session("abc123")
print(f"Messages: {len(details['messages'])}")

delete_session()

client.delete_session(session_id: str) -> None

Delete a session.Example:

client.delete_session("abc123")

add_message()

client.add_message(
    session_id: str,
    role: str,
    content: str | None = None,
    parts: list[dict] | None = None
) -> Dict[str, Any]

Add message to session.Parameters:

role: "user" or "assistant"
content: Text content (simple mode)
parts: Parts array for multimodal (TextPart, ContextPart, ToolPart)

Example:

# Simple text message
client.add_message(
    session_id="abc123",
    role="user",
    content="Hello"
)

# With parts
client.add_message(
    session_id="abc123",
    role="assistant",
    parts=[
        {"type": "text", "text": "Response"},
        {"type": "context", "uri": "viking://..."}
    ]
)

commit_session()

client.commit_session(session_id: str) -> Dict[str, Any]

Commit session (archive and extract memories).Example:

result = client.commit_session("abc123")
print(f"Memories extracted: {result['memory_count']}")

Relations

link()

client.link(
    from_uri: str,
    uris: str | List[str],
    reason: str = ""
) -> None

Create relation link(s).Example:

# Single link
client.link(
    "viking://resources/doc1.md",
    "viking://resources/doc2.md",
    reason="Related documentation"
)

# Multiple links
client.link(
    "viking://resources/index.md",
    [
        "viking://resources/guide1.md",
        "viking://resources/guide2.md"
    ],
    reason="Index references"
)

unlink()

client.unlink(
    from_uri: str,
    uri: str
) -> None

Remove relation link.Example:

client.unlink(
    "viking://resources/doc1.md",
    "viking://resources/doc2.md"
)

relations()

client.relations(uri: str) -> List[Dict[str, Any]]

Get all relations for a resource.Returns:

[
    {"uri": "viking://...", "reason": "Related doc"},
    {"uri": "viking://...", "reason": "Reference"}
]

Example:

rels = client.relations("viking://resources/doc.md")
for rel in rels:
    print(f"{rel['uri']}: {rel['reason']}")

Import/Export

export_ovpack()

client.export_ovpack(
    uri: str,
    to: str
) -> str

Export resource as .ovpack file.Example:

output_path = client.export_ovpack(
    uri="viking://resources/docs",
    to="./backups/docs.ovpack"
)
print(f"Exported to: {output_path}")

import_ovpack()

client.import_ovpack(
    file_path: str,
    target: str,
    force: bool = False,
    vectorize: bool = True
) -> str

Import .ovpack file.Example:

root_uri = client.import_ovpack(
    file_path="./backups/docs.ovpack",
    target="viking://resources/restored",
    vectorize=True
)
print(f"Imported to: {root_uri}")

Health and Status

get_status()

client.get_status() -> Dict[str, Any]

Get system status.Example:

status = client.get_status()
print(f"Healthy: {status['is_healthy']}")
for name, info in status['components'].items():
    print(f"{name}: {info['is_healthy']}")

is_healthy()

client.is_healthy() -> bool

Quick health check.Example:

if client.is_healthy():
    print("System OK")
else:
    print("System has issues")

observer

client.observer -> Observer

Access component status.Example:

queue = client.observer.queue
print(f"Queue healthy: {queue['is_healthy']}")
print(f"Queue size: {queue.get('queue_size', 0)}")

vectordb = client.observer.vikingdb
vlm = client.observer.vlm
system = client.observer.system

Complete Example

import openviking as ov
from openviking_cli.utils import run_async

# Initialize client
client = ov.SyncHTTPClient(
    url="http://localhost:1933",
    api_key="your-key",
    agent_id="demo-agent"
)

try:
    client.initialize()
    
    # Add resource
    result = client.add_resource(
        path="https://github.com/volcengine/OpenViking/README.md",
        reason="Project documentation"
    )
    root_uri = result["root_uri"]
    
    # Wait for processing
    client.wait_processed(timeout=120)
    
    # Read content layers
    abstract = client.abstract(root_uri)  # L0
    overview = client.overview(root_uri)  # L1
    content = client.read(root_uri)        # L2
    
    # Semantic search
    results = client.find(
        "how to install openviking",
        limit=5,
        score_threshold=0.7
    )
    
    for r in results.resources:
        print(f"{r.uri} (score: {r.score:.4f})")
    
    # Session workflow
    session = client.session()
    run_async(session.add_message(role="user", content="Tell me about OpenViking"))
    run_async(session.add_message(role="assistant", content="OpenViking is..."))
    
    # Context-aware search
    ctx_results = client.search(
        "how to use it",
        session=session,
        limit=3
    )
    
    # Commit session
    run_async(session.commit())
    
    # Check health
    if client.is_healthy():
        print("System healthy")
    
finally:
    client.close()

Configuration

Client configuration and setup

CLI Usage

Command-line interface guide

Authentication

API key setup for HTTP mode

Deployment

Deploy OpenViking Server

Get Started

Core Concepts

Guides

Integrations

Installation

Client Types

Quick Start

Client Initialization

SyncOpenViking / AsyncOpenViking

SyncHTTPClient / AsyncHTTPClient

Core Methods

Lifecycle

Resource Management

Search and Retrieval

File System Operations

Content Access

Session Management

Relations

Import/Export

Health and Status

Complete Example

Configuration

CLI Usage

Authentication

Deployment

​Installation

​Client Types

​Quick Start

​Client Initialization

​SyncOpenViking / AsyncOpenViking

​SyncHTTPClient / AsyncHTTPClient

​Core Methods

​Lifecycle

​Resource Management

​Search and Retrieval

​File System Operations

​Content Access

​Session Management

​Relations

​Import/Export

​Health and Status

​Complete Example

​Related Resources

Configuration

CLI Usage

Authentication

Deployment

Installation

Client Types

Quick Start

Client Initialization

SyncOpenViking / AsyncOpenViking

SyncHTTPClient / AsyncHTTPClient

Core Methods

Lifecycle

Resource Management

Search and Retrieval

File System Operations

Content Access

Session Management

Relations

Import/Export

Health and Status

Complete Example

Related Resources