hikigai-appsdk

Python SDK for invoking AI agents in your applications.

Installation

Note: The Hikigai SDKs are currently published to Test PyPI while in early access. Do not run the SDK from the repository source directly — the hikigai.core namespace package must be installed to resolve correctly. Always install via the commands below.

# Using pip
pip install \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  hikigai-appsdk==0.0.1

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

Once released on the main PyPI index, installation will simplify to:

# Future release (main PyPI)
pip install hikigai-appsdk

Quick Start

Using the SDK (Recommended)

from hikigai.appsdk import AppClient

# Initialize client
client = AppClient(
    api_key="your-api-key",
    project_id="your-project-id"
)

# 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

Using Direct API (Advanced)

For custom integrations or when you need direct HTTP control:

Step 1: Exchange API Key for Session Token

curl --location --request POST 'http://localhost:8000/api/v1/auth/exchange' \
--header 'X-API-Key: your_api_key_here' \
--header 'Content-Type: application/json'

Step 2: Invoke Agent via API

import requests
import os
from datetime import datetime, timedelta

class HikigaiAPIClient:
    def __init__(self, api_key: str, base_url: str = "http://localhost:8000"):
        self.api_key = api_key
        self.base_url = base_url
        self.access_token = None
        self.token_expiry = None
    
    def exchange_api_key(self):
        """Exchange API Key for session token."""
        headers = {"X-API-Key": self.api_key, "Content-Type": "application/json"}
        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"]
        self.token_expiry = datetime.utcnow() + timedelta(seconds=data["expires_in"] - 300)
        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()
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }
        payload = {
            "message": message,
            "stream": False,
        }
        if session_id:
            payload["session_id"] = session_id
        
        response = requests.post(
            f"{self.base_url}/api/v1/agents/{agent_id}/invoke",
            headers=headers,
            json=payload
        )
        response.raise_for_status()
        return response.json()

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

# Invoke agent
with HikigaiAPIClientContext(api_key=os.environ["HIKIGAI_API_KEY"]) 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.

Platform Events

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

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.

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

Documentation

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

License

MIT