hikigai-agentsdk

Python SDK for deploying and managing AI agents on the Hikigai platform.

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-agentsdk==0.0.1

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

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

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

Quick Start

from hikigai.agentsdk import AgentClient, AgentConfig, InputSchema, OutputSchema, StringField

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

# Deploy an agent
agent = client.deploy(AgentConfig(
    name="my-agent",
    display_name="My AI Agent",
    description="A helpful AI assistant",
    instruction="You are a helpful assistant that answers questions",
    model="claude-3.5-sonnet",
    input_schema=InputSchema(fields={"message": StringField(required=True)}),
    output_schema=OutputSchema(fields={"response": StringField()}),
    tools=[],
    timeout=60,
    memory_mb=512,
    version="1.0.0"
))

print(f"Agent deployed! ID: {agent.id}")

Direct API Methods (Alternative Approach)

For custom integrations or when you need direct HTTP control, you can call the API endpoints directly:

Authentication Flow

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: Deploy Agent via API

Python Example:

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 deploy_agent(self, config: dict) -> dict:
        """Deploy an agent via API."""
        self.ensure_token_valid()
        headers = {
            "Authorization": f"Bearer {self.access_token}",
            "Content-Type": "application/json"
        }
        response = requests.post(
            f"{self.base_url}/api/v1/agents/deploy",
            headers=headers,
            json=config
        )
        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

# Deploy agent
with HikigaiAPIClientContext(api_key=os.environ["HIKIGAI_API_KEY"]) as client:
    config = {
        "name": "medical-coder",
        "display_name": "Medical Coding Assistant",
        "description": "Extracts ICD-10 and CPT codes",
        "instruction": "You are a medical coding expert...",
        "model": "claude-3.5-sonnet",
        "timeout": 60,
        "memory_mb": 512,
        "input_schema": {
            "fields": {
                "clinical_note": {"type": "string", "required": True}
            }
        },
        "output_schema": {
            "fields": {
                "icd10_codes": {"type": "array"},
                "cpt_codes": {"type": "array"}
            }
        },
        "tools": []
    }
    
    result = client.deploy_agent(config)
    print(f"Agent deployed! ID: {result['agent_id']}")

For complete API documentation, see the Python AgentSDK Docs.

Platform Events

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

sub = client.events.create_webhook(
    url="https://ci.example.com/hooks/hikigai",
    event_types=["agent.deployed", "agent.deleted"],
)
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.agentsdk 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-agentsdk[live]'):

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

See the Platform Events guide.

Rooms & WebRTC

Join project-scoped rooms over WebSocket for pub/sub fan-out, and use the same socket as the signalling channel for peer-to-peer WebRTC calls (pip install 'hikigai-agentsdk[live]'):

from hikigai.agentsdk import RoomSession

async with RoomSession(base_url, api_key=KEY) as room:
    await room.join("call-1")
    await room.rtc_join("call-1")            # requires the "rtc" capability
    async for frame in room:
        if frame["type"] == "rtc_peer_joined" and frame["initiator"]:
            await room.rtc_offer("call-1", frame["id"], await make_offer())
        elif frame["type"] == "rtc_offer":
            await room.rtc_answer("call-1", frame["sender"], await answer(frame["sdp"]))

The SDK carries signalling only (SDP + ICE) — pair it with your own media stack (e.g. aiortc). Media flows directly between peers, never through the platform. See the Rooms & WebRTC signalling section of the SDK docs.

Features

• 🚀 Simple Deployment: Deploy agents with a single function call
• 🔧 Tool Integration: Add custom tools, OpenAPI specs, or MCP servers
• 📊 Schema Validation: Define input/output schemas for type safety
• ☁️ Multi-Cloud: Deploy to GCP Cloud Run, GCP Agent Engine, or AWS Bedrock
• 🔒 HIPAA Compliant: Built-in compliance checking
• 📈 Versioning: Semantic versioning support
• 🔔 Platform Events: Signed webhooks + live event stream via client.events
• 📞 Rooms & WebRTC: WebSocket pub/sub rooms + P2P call signalling via RoomSession

Documentation

Full documentation:

License

MIT