Python SDK

vedika-agent-sdk is the typed async Python client for the Vedika agent harness. Same wire, same chunk allowlist, swappable transport.

Install

pip install vedika-agent-sdk

Version 0.2.0 on PyPI. Python 3.10 or newer.

Quickstart

import asyncio
from vedika_agent import create_agent

async def main():
    agent = create_agent("https://harness.example.com", api_key="...")
    thread = await agent.threads.create()
    turn = await agent.threads.send_turn(thread.id, {
        "kind": "text", "text": "Show my kundli for 1990-06-15.",
    })
    async for chunk in turn:
        if chunk.type == "text-delta":
            print(chunk.delta, end="")
    await turn.done()

asyncio.run(main())

API

create_agent(base_url, api_key, *, default_headers=None, transport=None)

Returns a VedikaAgent with a .threads namespace.

await agent.threads.create(opts=None)

POST /api/v1/agent/threads. Returns a Thread.

await agent.threads.send_turn(thread_id, input, opts=None)

POST /threads/{id}/turns. Returns a Turn.

from vedika_agent import create_agent, MemoryToggles

agent = create_agent("https://harness.example.com", api_key="...")
turn = await agent.threads.send_turn(
    thread.id,
    {"kind": "text", "text": "Career outlook for 2026?"},
    model="vedika-swift",
    memory=MemoryToggles(old_memory=True),
)

await agent.threads.resume_turn(turn_id, *, last_event_id)

GET /turns/{id}/events with Last-Event-ID.

await agent.threads.cancel_turn(turn_id)

POST /turns/{id}/cancel. Idempotent.

The Turn object

Memory toggles

Pass memory=MemoryToggles(old_memory=..., sensitive_info=...) to send_turn. They are sent as x-vedika-memory-old and x-vedika-memory-sensitive. Both default to False.

Errors

See Errors. The SDK exposes VedikaAgentError, AuthenticationError, NotFoundError, RateLimitError (with retry_after_seconds), InvalidRequestError, ServerError, TurnCancelledError, StreamError, and ProtocolError. Use is_vedika_agent_error(e) to discriminate.

Injecting a transport

Tests inject a Transport whose request method returns an object with status, headers, body, iter_bytes(), ok, text(), and json(). The default uses httpx.AsyncClient.

class FakeTransport:
    async def request(self, method, url, *, headers, body=None, stream=False):
        # return a stub _HttpResponse that yields bytes
        ...

agent = create_agent("https://harness.example.com", api_key="...", transport=FakeTransport())

Reserved headers

The SDK refuses to override authorization, accept, content-type, idempotency-key, last-event-id, and x-vercel-ai-ui-message-stream.