Application SDK

import asyncio
from hikigai.appsdk import AppClient

# Initialize client (or set HIKIGAI_API_KEY / HIKIGAI_PROJECT_ID)
client = AppClient(
    api_key="your-api-key",
    project_id="your-project-id",
)

# Get an agent by slug or id
agent = client.agent("medical-coder")

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

# Stream
for chunk in agent.stream("Tell me about diabetes"):
    print(chunk, end="", flush=True)

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

# Live WebSocket (install: pip install websockets)
async def live():
    async with client.live(agent.slug, user_id="app-user") as session:
        await session.send_text("Hello")
        async for event in session:
            print(event)
            if event.get("type") in ("turn_complete", "end"):
                break

asyncio.run(live())

from hikigai.appsdk import AppClient

client = AppClient(api_key="...", project_id="...")
agent = client.agent("vitals-fetch-subagent")

# Same connectors shape for simple LLM roots and Sequential leaves
response = agent.invoke(
    {"site_id": 1, "patient_id": "PATIENT001"},
    connectors={"idexx-mcp-server": {}},
    timeout=120,
)

print(response.content)                      # summary text / {"response": "..."}
print(response.metadata.tools_called)        # e.g. ["get_vitals_for_today"]
print(response.metadata.latency_ms)

# Live WebSocket — MCP on leaves too
# async with client.live(
#     "vitals-fetch-subagent",
#     connectors={"idexx-mcp-server": {}},
# ) as session:
#     await session.start(modality="text")
#     await session.send_text("Fetch today's vitals")
#     async for event in session:
#         if event.get("type") == "tool_call":
#             print(event)  # leaf CallTool frames

import asyncio
from hikigai.appsdk import AppClient

async def main():
    client = AppClient(api_key="...", project_id="...")
    client.get_auth_token()  # POST /auth/exchange — cache for mint
    # AppSDK generates session_id if omitted; always mints with {session_id, user_id}

    async with client.live("YOUR_AGENT_SLUG", user_id="app-transcriber") as session:
        await session.start(modality="audio", mime_type="audio/pcm;rate=16000")
        await session.send_data(pcm_bytes)  # raw bytes → binary frames

        async for event in session:
            if event.get("source") == "input_transcription":
                print(event["content"])  # replace UI on partials
            if event.get("type") == "end":
                break

        await session.end()

asyncio.run(main())

async with client.live(
    "clinical-live-agent",
    user_id="app-user",
    connectors={"epic-ehr": {}},  # same shape as invoke
) as session:
    await session.send_text("Look up the patient chart")
    async for event in session:
        if event.get("type") == "tool_call":
            print(event)
        if event.get("type") in ("turn_complete", "end"):
            break

import asyncio
from hikigai.appsdk import AppClient

async def main():
    client = AppClient(api_key="hikigai_...", project_id="...")

    session = await client.speech.translate(
        target_language="es",
        echo_target_language=True,   # default: echo when input is already Spanish
        input_transcription=True,    # default
        output_transcription=True,   # default
    )

    async def send():
        # Any source format — converted to 16 kHz mono PCM for you.
        async for chunk in phone_call():
            await session.send_audio(chunk, encoding="mulaw", sample_rate=8000)

    asyncio.create_task(send())

    async for event in session:
        if event.type == "audio":
            speaker.write(event.audio)          # 24 kHz PCM
        elif event.type == "transcript":
            print(event.source, event.text, event.language_code)     # "input" / "output"

asyncio.run(main())

session = await client.speech.transcribe(
    languages=["en-US", "es-ES"],               # optional; omit to auto-detect
    custom_vocabulary=["Hikigai", "Kubernetes"],
)

await session.send_audio(pcm_16k)               # already 16 kHz mono PCM

async for event in session:
    if event.type == "transcript":
        print(event.text, event.language_code, "final" if event.final else "interim")

await session.set_language("fr")   # no reconnect, no new token

from hikigai.appsdk import to_model_audio, chunk_audio, pcm_to_wav

pcm = to_model_audio(raw, encoding="mulaw", sample_rate=8000)  # telephony → 16 kHz
pcm = to_model_audio(raw, encoding="f32", sample_rate=48000)   # browser Float32
pcm = to_model_audio(raw, sample_rate=48000, channels=2)       # stereo → mono

for chunk in chunk_audio(pcm):        # ~100 ms frames, never splits a sample
    await session.send_audio(chunk)

open("out.wav", "wb").write(pcm_to_wav(translated))  # 24 kHz by default

token = client.speech.mint_token(
    "translate",
    target_language_code="ja",
    locked=True,   # pin the config — client cannot call configure
)
# Hand token["speech_token"] to the browser; it connects directly.

# Create a context store
ctx = client.create_session(external_user_ref="dr-smith", ttl_seconds=3600)
ctx.set("patient_id", "MRN-12345")
ctx.set("visit_id", "V-99", if_match=None)

print(ctx.get("patient_id"))
print(ctx.get_all())
print(ctx.metadata())

# Attach later
ctx = client.session(ctx.session_id)
ctx.update({"chief_complaint": "chest pain"})
ctx.touch(ttl_seconds=7200)
ctx.delete("visit_id")
ctx.clear()

import asyncio
from hikigai.appsdk import RoomSession

async def main():
    room = RoomSession(
        base_url="https://backend.hikigaiplatform.io",
        api_key="your-api-key",
        project_id="your-project-id",
    )
    async with room:
        await room.join("care-team-7")  # creates the room if new
        await room.publish("care-team-7", "note", {"text": "Labs ready"})
        await room.who("care-team-7")
        async for frame in room:
            if frame.get("type") == "message":
                print(frame["event"], frame["data"])
            elif frame.get("type") == "error":
                print(frame.get("code"), frame.get("message"))
                break

asyncio.run(main())

await room.join("call-1")
await room.rtc_join("call-1")
async for frame in room:
    if frame["type"] == "rtc_peer_joined" and frame.get("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"]))

from hikigai.appsdk import (
    validate_even_g2_manifest,  # same app.json rules as the deploy pipeline
    paginate_for_glasses,       # split agent answers into display-sized pages
    normalize_for_glasses,      # strip markdown decorations agents emit
    even_g2_dev_mode_url,       # validate the live URL used in the dev-mode QR
    EVEN_G2_DISPLAY,            # {"width": 640, "height": 350}
    EVEN_G2_MAX_TEXT_CHARS,     # 2000 (official text-heavy template cap)
)

check = validate_even_g2_manifest(json.load(open("app.json")))
if not check.valid:
    raise SystemExit(check.error)

pages = paginate_for_glasses(response.content, max_chars=420)
qr_payload = even_g2_dev_mode_url(live_url)

import os
from fastapi import FastAPI
from pydantic import BaseModel
from hikigai.appsdk import AppClient, normalize_for_glasses, paginate_for_glasses

app = FastAPI()
client = AppClient()  # reads HIKIGAI_API_KEY / HIKIGAI_PROJECT_ID
PAGE_CHARS = int(os.environ.get("GLASSES_PAGE_CHARS", "420"))
AGENT_ID = os.environ["HIKIGAI_AGENT_ID"]

class AskRequest(BaseModel):
    question: str
    sessionId: str | None = None

@app.post("/api/ask")
async def ask(req: AskRequest):
    agent = client.agent(AGENT_ID)
    response = agent.invoke(req.question, session_id=req.sessionId)

    clean = normalize_for_glasses(response.content)
    return {
        "answer": response.content,
        "pages": paginate_for_glasses(clean, max_chars=PAGE_CHARS),
        "sessionId": response.session_id,
    }

class AudioRequest(BaseModel):
    audio_base64: str
    mime_type: str | None = "audio/wav"
    sessionId: str | None = None

@app.post("/api/ask-audio")
async def ask_audio(req: AudioRequest):
    agent = client.agent(AGENT_ID)
    response = agent.invoke(
        "Answer the question in this recording.",
        session_id=req.sessionId,
        attachments=[{
            "data_base64": req.audio_base64,
            "mime_type": req.mime_type or "audio/wav",
        }],
    )

    clean = normalize_for_glasses(response.content)
    return {
        "pages": paginate_for_glasses(clean, max_chars=PAGE_CHARS),
        "sessionId": response.session_id,
    }

from hikigai.appsdk import AppClient

client = AppClient(api_key=..., project_id=...)

catalog = client.cloud.catalog(workload="app")
for service in catalog.services:
    print(service.id, service.display_name, service.regions[0].id)

result = client.cloud.validate_target(
    workload="app",
    service_id="gcp-cloud-run",
    region="europe-west1",
    config={"cpu": "2", "memory": "1Gi"},
)

sub = client.events.create_webhook(
    url="https://ci.example.com/hooks/hikigai",
    event_types=["agent.deployed", "agent.deleted"],
    description="deploy notifications",
)
print(sub.secret)  # shown once — store it

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

with open("audio.wav", "rb") as f:
    uploaded = client.storage.upload(
        app_id="app_123",
        file=f,
        filename="audio.wav",
        content_type="audio/wav",
        metadata={"session": "s-9"},
        ttl=86400,
    )

url = client.storage.signed_url("app_123", uploaded.object_id, ttl=3600)
print(url.signed_url, url.expires_at)

info = client.storage.get("app_123", uploaded.object_id)
print(info.original_filename, info.size_bytes, info.metadata)

page = client.storage.list("app_123", content_type="audio/wav", status="active", limit=20, page=1)
for obj in page.objects:
    print(obj.object_id, obj.original_filename)
print(page.total, page.total_pages)

client.storage.delete("app_123", uploaded.object_id)

response = agent.invoke(
    "Draft discharge summary",
    plugin_context={"sona": {"user_id": "dr-smith-uuid"}},
)
output_id = response.plugins["sona"]["output_id"]

client.sona.submit_edit(output_id, "User-edited note text...")
client.sona.approve_output(output_id, user_id="dr-smith-uuid")
prefs = client.sona.get_preferences("dr-smith-uuid", agent_id=agent.id)
suggestions = client.sona.get_suggestions("dr-smith-uuid")