Application SDK

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
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 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 session
import asyncio
async def live():
    async with agent.live_session() as session:
        await session.connect()
        await session.send_text("Hello")
        async for event in session:
            print(event)
asyncio.run(live())

import asyncio
from hikigai.appsdk import AppClient

async def main():
    client = AppClient(api_key="...", project_id="...")
    agent = client.agent("live-scribe")

    async with agent.live_session() as session:
        await session.connect()
        await session.start(modality="text")  # or "audio"
        await session.send_text("Patient presents with fever")
        # await session.send_audio(pcm_bytes, mime_type="audio/pcm;rate=16000")

        async for event in session:
            print(event)
            if event.get("type") == "turn_complete":
                break

        await session.end()
        await session.close()

asyncio.run(main())

# 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://api.hikigai.com",
        api_key="your-api-key",
        project_id="your-project-id",
    )
    async with room:
        await room.join("care-team-7")
        await room.publish("care-team-7", "note", {"text": "Labs ready"})
        await room.who("care-team-7")
        async for frame in room:
            print(frame)

asyncio.run(main())

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)

uploaded = client.storage.upload(
    app_id="app_123",
    file=open("audio.wav", "rb"),
    filename="audio.wav",
    content_type="audio/wav",
)
url = client.storage.signed_url("app_123", uploaded.object_id, ttl=3600)
client.storage.list("app_123")
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")