Developer Guide

Overview

Getting Started

# Your token looks like this:
hk_token_abc123def456ghi789...

pip install hikigai-agentsdk hikigai-appsdk hikigai-core --upgrade

export HIKIGAI_API_KEY=hk_token_your_token

from hikigai import Hikigai
client = Hikigai(api_key="hk_token_your_token")

Deploy Your First Agent

Create a simple sentiment analyzer in just a few lines of code:

from hikigai import Hikigai, AgentConfig

# Initialize client
client = Hikigai()

# Create and deploy agent
agent = client.deploy(AgentConfig(
    name="sentiment-analyzer",
    instruction="""
    You are a sentiment analysis expert.
    Analyze the sentiment of the given text and respond with:
    - "positive" for positive sentiment
    - "negative" for negative sentiment  
    - "neutral" for neutral sentiment
    Also provide a confidence score from 0 to 1.
    Format: {"sentiment": "...", "confidence": 0.XX}
    """,
    model="gemini-2.0-flash",
    description="Analyzes text sentiment",
    category="NLP",
))

print(f"✓ Agent deployed: {agent.id}")
print(f"✓ Endpoint: {agent.endpoint}")

Use Your Agent

Basic Invocation

# Get your deployed agent
agent = client.agent("sentiment-analyzer")

# Invoke it!
response = agent.invoke("I absolutely love this product!")
print(response.content)
# Output: {"sentiment": "positive", "confidence": 0.95}

With Conversation Memory

# Use session_id to maintain conversation context
session = "user-abc-123"

r1 = agent.invoke("Remember: my favorite color is blue", session_id=session)
r2 = agent.invoke("What's my favorite color?", session_id=session)
print(r2.content)  # "Your favorite color is blue"

Streaming Responses

# Stream for real-time output
for chunk in agent.stream("Write a poem about AI"):
    print(chunk, end="", flush=True)

Using cURL

curl -X POST "https://api.hikigai.com/v1/agents/sentiment-analyzer/invoke" \
  -H "Authorization: Bearer hk_token_your_token" \
  -H "Content-Type: application/json" \
  -d '{"message": "I love this!"}'

Project Collaboration

Hikigai is built for teams. You can invite collaborators to your projects and assign them specific roles.

Bring Your Own Key (BYOK)

For production workloads, you can provide your own LLM provider API keys. This gives you more control over rate limits and direct billing with providers.

MCP Servers & Connectors

Connect agents to clinical data via the Model Context Protocol (MCP). Use registry connectors (Epic EHR, Cerner, Athena Health, Google Health Connect, …) or deploy your own MCP server from GitHub in the console.

Using a connector at invoke time

from hikigai.appsdk import AppClient

client = AppClient(api_key="...", project_id="...")
result = client.agent("clinical-note-coder").invoke(
    {"note": "..."},
    connectors={"epic-ehr": {}},
)

Multi-Cloud Deployment

Agents, apps, and MCP servers deploy to GCP or AWS. Optional BYOC credentials deploy into your own account.

In the console, pick provider → service → region (and zones where supported) during deploy. From the SDK, pass cloud_service / region on AgentSDK deploy(), or use client.cloud.catalog() on AppSDK.

Rooms & Event Bus

Rooms provide project-scoped WebSocket pub/sub (presence, publish, bounded replay) via SDK RoomSession clients. See API reference — Rooms.

The Event Bus delivers platform CloudEvents (agent deploy/delete, jobs, invocations, storage). Subscribe with project webhooks (HMAC-signed) or client.events.stream(). Configure webhooks in the project console.

Production Tips

Example: Production-Ready Code

import asyncio
from hikigai import HikigaiAsync, AgentConfig
from hikigai.exceptions import RateLimitError, HikigaiError

async def invoke_with_retry(agent, message, max_retries=3):
    """Invoke with exponential backoff retry."""
    for attempt in range(max_retries):
        try:
            return await agent.invoke(message)
        except RateLimitError as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(e.retry_after)
            else:
                raise
        except HikigaiError as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
            else:
                raise

async def main():
    async with HikigaiAsync() as client:
        agent = await client.agent("my-agent")
        response = await invoke_with_retry(agent, "Hello!")
        return response.content

asyncio.run(main())

Framework Support

Hikigai is designed to be framework-agnostic. Under the hood, we support:

Best Practices