hikigai-appsdk

Python SDK for invoking AI agents in your applications.

Installation

pip install hikigai-appsdk

This will automatically install hikigai-core (the shared namespace dependency) alongside hikigai-appsdk.

Pin a version if needed:

pip install hikigai-appsdk==0.2.7

Quick Start

Using the SDK (Recommended)

from hikigai.appsdk import AppClient

# Initialize client
client = AppClient()  # your computer: your `hikigai login`; deployed App: the injected key

# Get an agent
agent = client.agent("medical-coder")

# Invoke the agent
response = agent.invoke("Patient presents with fever and cough...")
print(response.content)

# Stream responses
for chunk in agent.stream("Tell me a story"):
    print(chunk, end="")

# Session-based conversations
session_agent = agent.with_session("user-123")
session_agent.invoke("What is diabetes?")
session_agent.invoke("How is it treated?")  # Remembers context

Authentication

There is nothing to paste. The same code runs on your computer and in your deployed App; the SDK picks the right credential:

On your computer: hikigai login

Installing this package gives you the hikigai command:

hikigai login                  # opens the browser: pick the project, check the code, Approve
hikigai login --project <id>   # or name the project here
hikigai whoami                 # who, which project, time left
hikigai token                  # print the token (curl / Postman: Authorization: Bearer <token>)
hikigai logout                 # end it now

The sign-in is saved in ~/.hikigai/credentials.json (readable only by you) and AppClient() with no arguments uses it. It works for one project, with your role in that project, for 6 hours: invoke, stream, live voice, rooms, speech, MCP, and deploy / manage if your role allows. See or end your sign-ins in the console under CLI logins. It does not work inside deployed Apps. To use a local platform: hikigai login --platform-url http://localhost:8000 and AppClient(base_url="http://localhost:8000").

Inside a deployed App

No credentials are needed. The platform injects HIKIGAI_API_KEY (a Deploy key, hikigai_deploy_...), HIKIGAI_PROJECT_ID, HIKIGAI_APP_ID, HIKIGAI_PLATFORM_URL (the platform helper inside the App; use it for all platform calls) and HIKIGAI_PUBLIC_PLATFORM_URL (the public address, for browsers) into the container, and the SDK reads them:

client = AppClient()  # reads the injected env vars

The SDK exchanges the Deploy key for a workload token (POST $HIKIGAI_PLATFORM_URL/api/v1/auth/exchange; the platform adds the cloud identity proof for you); the token lasts 5 minutes and is renewed automatically. The SDK caches it and retries once on 401.

Order: explicit api_key= / project_id= / base_url= arguments, then the HIKIGAI_* environment variables, then your hikigai login. Project API keys were removed: an old one gets PROJECT_KEYS_REMOVED.

Errors from the platform carry a machine-readable code on the exception (err.code) and a readable err.message: DEPLOY_KEY_REQUIRED, INVALID_KEY and CLI_SESSION_INVALID (login ended or revoked: run hikigai login) raise AuthenticationError (401); KEY_PROJECT_MISMATCH, CLI_PROJECT_MISMATCH, CLI_SESSION_NOT_ALLOWED, APP_DEPLOY_REQUIRED, APP_STOPPED, REQUIRES_ATTESTATION, PLATFORM_NETWORK_REQUIRED and DEPLOY_KEY_NOT_ALLOWED raise PermissionDeniedError (403); RATE_LIMITED raises RateLimitError (429).

Using Direct API (Advanced)

For custom integrations or when you need direct HTTP control. These calls run inside your deployed App, using the env vars the platform sets.

Step 1: Exchange the Deploy Key for an Access Token

# Inside the deployed App. No body. Returns access_token, expires_in 300 (5 minutes).
curl -X POST "$HIKIGAI_PLATFORM_URL/api/v1/auth/exchange" \
  -H "X-API-Key: $HIKIGAI_API_KEY" \
  -H "X-Project-ID: $HIKIGAI_PROJECT_ID"

Step 2: Invoke Agent via API

import requests
import os
from datetime import datetime, timedelta

class HikigaiAPIClient:
    def __init__(self, api_key: str, project_id: str, base_url: str = os.environ["HIKIGAI_PLATFORM_URL"]):
        self.api_key = api_key
        self.project_id = project_id
        self.base_url = base_url
        self.access_token = None
        self.token_expiry = None
    
    def exchange_api_key(self):
        """Exchange the Deploy key for an access token (no body)."""
        headers = {"X-API-Key": self.api_key, "X-Project-ID": self.project_id}
        response = requests.post(
            f"{self.base_url}/api/v1/auth/exchange",
            headers=headers
        )
        response.raise_for_status()
        data = response.json()
        
        self.access_token = data["access_token"]
        # renew at 80% of the lifetime
        self.token_expiry = datetime.utcnow() + timedelta(seconds=data["expires_in"] * 0.8)
        return data
    
    def ensure_token_valid(self):
        """Refresh token if needed."""
        if not self.access_token or datetime.utcnow() >= self.token_expiry:
            self.exchange_api_key()
    
    def invoke_agent(self, agent_id: str, message: str, session_id: str = None) -> dict:
        """Invoke an agent via API."""
        self.ensure_token_valid()
        payload = {
            "message": message,
            "stream": False,
        }
        if session_id:
            payload["session_id"] = session_id
        
        def post():
            headers = {
                "Authorization": f"Bearer {self.access_token}",
                "X-Project-ID": self.project_id,
                "Content-Type": "application/json"
            }
            return requests.post(
                f"{self.base_url}/api/v1/agents/{agent_id}/invoke",
                headers=headers,
                json=payload
            )
        
        response = post()
        if response.status_code == 401:  # token expired: exchange again once
            self.exchange_api_key()
            response = post()
        response.raise_for_status()
        return response.json()

# Usage with context manager
class HikigaiAPIClientContext:
    def __init__(self, api_key: str, project_id: str):
        self.client = HikigaiAPIClient(api_key, project_id)
    
    def __enter__(self):
        self.client.exchange_api_key()
        return self.client
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        pass

# Invoke agent (inside the deployed App)
with HikigaiAPIClientContext(
    api_key=os.environ["HIKIGAI_API_KEY"],        # Deploy key set by the platform
    project_id=os.environ["HIKIGAI_PROJECT_ID"],
) as client:
    result = client.invoke_agent(
        agent_id="medical-coder",
        message="Patient presents with fever and cough...",
        session_id="user-123"
    )
    print(result["content"])

For complete API documentation, see the Python AppSDK Docs.

Call an MCP Server Directly

Agents use MCP servers for you (pass connectors). To call one of your MCP servers yourself, ask Hikigai for an MCP token: a ticket that works for 1 hour, for that one server only. AppClient signs in for you (the App's key in a deployed App, your hikigai login on your computer).

# pip install mcp
from hikigai.appsdk import AppClient
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

client = AppClient()

# 1. MCP token for one server -> {"token", "header", "mcp_url"}
ticket = client.api.post("/api/v1/mcp/access-token", json={"server": "my-mcp-server"})

# 2. Call the MCP server with it
async with streamablehttp_client(ticket["mcp_url"], headers={ticket["header"]: ticket["token"]}) as (read, write, _):
    async with ClientSession(read, write) as mcp:
        await mcp.initialize()
        tools = await mcp.list_tools()

Without the SDK, inside the App: POST $HIKIGAI_PLATFORM_URL/api/v1/auth/exchange (App's token), then POST $HIKIGAI_PLATFORM_URL/api/v1/mcp/access-token with Authorization: Bearer <access_token>, then call mcp_url with X-Hikigai-Mcp-Token: <token>. On your computer, use $(hikigai token) as the bearer token and skip the exchange. When the MCP server answers 401, get a new MCP token.

Platform Events

React to platform facts (job.*, agent.deployed, invocation.completed, storage.*) instead of polling. Register a signed webhook, or open a live stream:

Runs inside a deployed App (AppClient() with no arguments), or on your computer after hikigai login.

sub = client.events.create_webhook(
    url="https://app.example.com/hooks/hikigai",
    event_types=["job.*", "agent.deployed"],
)
print(sub.secret)   # shown ONCE — store it in your secret manager

Verify every delivery before trusting it — over the raw body, since re-serializing parsed JSON will not reproduce the signed bytes:

from hikigai.appsdk import parse_webhook_event, SignatureVerificationError

try:
    event = parse_webhook_event(
        secret=os.environ["HIKIGAI_WEBHOOK_SECRET"],
        signature_header=request.headers["X-Hikigai-Signature"],
        body=request.get_data(as_text=True),
    )
except SignatureVerificationError:
    return "", 400

Deliveries are at-least-once, so dedupe on event["id"]. For a live subscription instead of an endpoint (pip install 'hikigai-appsdk[live]'):

async with client.events.stream(patterns=["job.*"]) as stream:
    async for event in stream:
        print(event["type"], event["data"])

See the Platform Events guide.

Rooms & WebRTC

SFU Mode (Recommended for multi-party calls)

Join rooms via HTTP with SFU (Selective Forwarding Unit) for scalable audio/video. No WebSocket required for media:

import os
from hikigai.appsdk.rooms import connect_media

# Get media token and ICE servers (inside the deployed App)
media = await connect_media(
    base_url=os.environ["HIKIGAI_PLATFORM_URL"],
    room="call-1",
    api_key=os.environ["HIKIGAI_API_KEY"],
    project_id=os.environ["HIKIGAI_PROJECT_ID"],
)

print(media["token"])      # JWT for hikigai-media
print(media["sfu_url"])    # platform-provided; don't hardcode
print(media["ice_servers"]) # STUN/TURN config

# Use with aiortc or browser WebRTC to join the SFU
# (for browsers, your App mints the media token and gives the browser only that token)
# Media flows: Browser ↔ hikigai-media (SFU) ↔ other participants

Benefits:

• Scales to 50+ participants
• Each participant uploads once, SFU fans out
• In-call chat via DataChannel (no WebSocket needed)
• HTTP-only join flow

Use RoomSession for WebSocket pub/sub (join/publish/who/presence) — not for media. For calls, use connect_media above.

See: FEATURES-WEBRTC-ROOMS.md for topology comparison and client examples.

Features

• 🚀 Simple Invocation: Call agents with a single method
• 📡 Streaming: Real-time streaming responses
• 💬 Sessions: Multi-turn conversations with context
• ⚡ Async Support: Built-in async/await support (coming soon)
• 🔍 Agent Discovery: List and search available agents
• 📊 Metadata: Access performance metrics and status
• 🔔 Platform Events: Signed webhooks + live event stream via client.events
• 📞 Rooms & WebRTC: WebSocket pub/sub rooms + SFU media via connect_media

Documentation

Full documentation: https://docs.hikigai.com/appsdk

License

MIT