AgentSDK

Agent Deployment SDK

Deploy and manage healthcare AI agents for medical coding, clinical documentation, and diagnosis assistance. Built for HIPAA-compliant medical applications with clinical schemas and medical tools. Perfect for healthcare ML engineers and medical agent developers.

v0.2.2Python 3.10+Type HintsHIPAA ReadyJobs & Rooms

Installation

Using pip

pip install hikigai-agentsdk

Quick Start

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

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

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

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

# Invoke / stream while developing
result = client.invoke(agent.id, "Hello")
for chunk in client.stream(agent.id, "Summarize this note"):
    print(chunk, end="")

Client Reference

class AgentClient

AgentClientCore

Main interface for agent deployment and management. Handles API communication, authentication, and agent lifecycle operations.

Constructor

client = AgentClient(
    api_key: Optional[str] = None,        # API key (or HIKIGAI_API_KEY env var)
    project_id: Optional[str] = None,     # Project ID (or HIKIGAI_PROJECT_ID env var)
    base_url: Optional[str] = None,       # API endpoint (defaults to production)
    timeout: float = 30.0,                # Request timeout in seconds
)

Methods

Method	Returns	Description
`deploy`(config, timeout=1200.0, poll_interval=10.0)	`DeployedAgent`	Deploy a new agent and poll until active
`deploy_from_file`(file_path, name=None, description=None, ...)	`DeployedAgent`	Deploy agent from a Python/ADK file
`list_agents`()	`List[DeployedAgent]`	List all deployed agents
`get_agent`(agent_id: str)	`DeployedAgent`	Get agent by ID or slug
`update_agent`(agent_id: str, update: Dict)	`DeployedAgent`	Update agent metadata/config and trigger redeploy
`delete_agent`(agent_id: str)	`None`	Delete an agent
`invoke`(agent_id, input, session_id=None, provider=None, model=None)	`Dict[str, Any]`	Invoke a deployed agent (dev/test helper)
`stream`(agent_id, input, session_id=None, provider=None, model=None)	`Iterator[str]`	SSE stream chunks from a deployed agent
`get_auth_token`()	`Dict[str, Any]`	Exchange API key for a short-lived access token

Properties

`jobs`()	`JobsClient`	Background job queue (enqueue, wait, stream, cancel, replay)
`storage`()	`StorageClient`	App object storage (upload, signed URLs, list, delete)

Context Manager

with AgentClient() as client:
    agent = client.deploy(config)
# Resources automatically released

class JobsClient (client.jobs)

client.jobsJobs

Background job queue for long-running agent work. Access via AgentClient.jobs.

Method	Returns	Description
`enqueue`(job_type, payload, agent_id=None, priority='standard', ...)	`EnqueuedJob`	Enqueue a background job
`get`(job_id: str)	`JobDetail`	Fetch job status and result
`list`(status=None, job_type=None, agent_id=None, page=1, per_page=20)	`List[JobDetail]`	List jobs with filters
`cancel`(job_id: str)	`JobDetail`	Cancel a queued or running job
`replay`(job_id, override_retry_policy=None, override_payload=None)	`EnqueuedJob`	Re-enqueue a failed job
`wait`(job_id, timeout=600.0, poll_interval=5.0)	`JobDetail`	Block until the job reaches a terminal status
`stream`(job_id: str)	`Iterator[Dict]`	Stream live job progress events

job = client.jobs.enqueue(
    "agent.invoke",
    {"input": "Batch-code these notes"},
    agent_id=agent.id,
    priority="standard",
)
detail = client.jobs.wait(job.id)
print(detail.status, detail.result)

Storage (client.storage)

client.storageStorage

Upload and manage application objects (charts, audio, attachments) with signed download URLs.

# Upload a file
uploaded = client.storage.upload(
    app_id="app_123",
    file=open("note.pdf", "rb"),
    filename="note.pdf",
    content_type="application/pdf",
)

# Signed download URL
url = client.storage.signed_url("app_123", uploaded.object_id, ttl=3600)
objects = client.storage.list("app_123")
client.storage.delete("app_123", uploaded.object_id)

class RoomSession

RoomSessionRealtime

Multi-party realtime rooms over WebSocket (presence, publish, sequence tracking). Construct directly — not attached to AgentClient.

import asyncio
from hikigai.agentsdk 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("rounds-42")
        await room.publish("rounds-42", "update", {"status": "in_progress"})
        async for frame in room:
            print(frame)

asyncio.run(main())

Configuration Reference

class AgentConfig

AgentConfigPydantic Model

Comprehensive configuration for deploying an agent. Supports schemas, tools, versioning, and HIPAA compliance.

Property	Type	Description
Identity
`name`*	`str`	Unique agent name (3-64 chars, lowercase, hyphens)
`display_name`*	`str`	Human-readable display name (3-100 chars)
`description`*	`str`	Short description (10-500 chars)
`long_description`	`str \| None`	Full documentation (max 5000 chars)
Core
`agent_type`	`str`	Agent orchestration type (llm, sequential, parallel, loop) Default: `llm`
`instruction`*	`str`	System prompt/instruction (10-50,000 chars)
`model`	`str`	LLM model to use Default: `claude-3.5-sonnet`
Multi-Agent Support
`sub_agents`	`List[SubAgentConfig]`	Sub-agents for workflow orchestration Default: `[]`
Advanced Features
`planner_config`	`PlannerConfig \| None`	Enable planning/reasoning capabilities
`code_execution`	`bool`	Enable code execution capability Default: `False`
`generation_config`	`GenerationConfig \| None`	LLM generation parameters
State Management
`output_key`	`str \| None`	Key to save agent output in state for downstream agents
`include_contents`	`str \| None`	Control conversation history inclusion
`max_iterations`	`int \| None`	Maximum iterations for LoopAgent
Classification
`category`	`str`	Category from predefined list Default: `documentation`
`tags`	`List[str]`	Tags for discovery (max 10) Default: `[]`
Schemas
`input_schema`*	`InputSchema \| Dict`	Input data structure definition
`output_schema`*	`OutputSchema \| Dict`	Output data structure definition
Modality
`input_modality`	`str`	Input modality: text, audio, or text_and_audio Default: `text`
`output_modality`	`str`	Output modality: text, audio, or text_and_audio Default: `text`
Tools & Connectors
`tools`	`List[Any]`	Tools available to the agent Default: `[]`
`connectors`	`List[ConnectorConfig]`	MCP connectors this agent uses Default: `[]`
Runtime
`timeout`	`int`	Request timeout in seconds (5-300) Default: `60`
`memory_mb`	`int`	Memory allocation in MB (128-4096) Default: `512`
`min_instances`	`int`	Minimum scaling instances (0-10) Default: `0`
`max_instances`	`int`	Maximum scaling instances (1-100) Default: `10`
Versioning
`version`*	`str`	Semantic version (e.g., '1.0.0')
`changelog`	`str \| None`	What changed in this version
Compliance & Visibility
`hipaa_compliant`	`bool`	Agent handles PHI per HIPAA Default: `True`
`public`	`bool`	Publicly listed in marketplace Default: `False`
Cloud Deployment
`cloud_provider`	`str`	Cloud provider Default: `gcp`
`gcp_region`	`str`	GCP region for deployment Default: `us-central1`
`aws_region`	`str`	AWS region for deployment Default: `us-east-1`
Healthcare Spec
`risk_tier`	`str \| None`	Healthcare risk tier Default: `low`
`clinical_domain`	`str \| None`	Clinical domain category
`a2a_skills`	`List[Dict]`	Machine-readable capabilities for AI-to-AI discovery Default: `[]`
`resource_limits`	`Dict[str, str]`	Container resource limits Default: `{'cpu': '1', 'memory': '512Mi'}`
`fhir_resources`	`List[Dict]`	Required FHIR resources for agent operation Default: `[]`
`prompt_version`	`str \| None`	Specific version of the clinical prompt Default: `1.0.0`
`requires_physician_oversight`	`bool`	Does this agent require human physician sign-off? Default: `False`
`fda_clearance_status`	`str`	FDA clearance status Default: `not-applicable`

Example

config = AgentConfig(
    name="medical-coder",
    display_name="Medical Coding Assistant",
    description="Extracts ICD-10 codes from clinical notes",
    instruction="You are a medical coding expert...",
    model="claude-3.5-sonnet",
    category="coding-billing",
    tags=["icd-10", "healthcare"],
    input_schema=InputSchema(fields={"note": StringField(required=True)}),
    output_schema=OutputSchema(fields={"codes": ArrayField()}),
    timeout=60,
    memory_mb=512,
    version="1.0.0",
    hipaa_compliant=True
)

Schema Types

InputSchema & OutputSchemaPydantic Model

Define the structure of data your agent expects and returns.

from hikigai.agentsdk import InputSchema, OutputSchema, StringField, IntegerField

input_schema = InputSchema(
    description="Patient information",
    fields={
        "patient_id": StringField(required=True),
        "age": IntegerField(required=True, minimum=0, maximum=150),
    }
)

output_schema = OutputSchema(
    description="Diagnosis results",
    fields={
        "diagnosis": StringField(),
        "confidence": IntegerField(minimum=0, maximum=100),
    }
)

Field Types

StringField

StringField(
    required=True,
    description="Field description",
    min_length=1,
    max_length=1000,
    pattern=r"^[A-Z0-9-]+$",
    enum=["option1", "option2"]
)

IntegerField

IntegerField(
    required=True,
    minimum=0,
    maximum=100,
    default=50
)

BooleanField

BooleanField(
    required=False,
    default=False
)

ArrayField

ArrayField(
    items=StringField(),
    min_items=1,
    max_items=10
)

ObjectField

ObjectField(
    properties={
        "name": StringField(),
        "value": IntegerField()
    },
    required_fields=["name"]
)

Models

class DeployedAgent

DeployedAgentPydantic Model

Represents a deployed agent with all deployment metadata returned from the API.

Property	Type	Description
`id`	`str`	Unique agent ID
`name`	`str`	Agent name (slug format)
`slug`	`str`	URL-friendly slug
`version`	`str`	Semantic version
`display_name`	`str \| None`	Human-readable name
`description`	`str \| None`	Short description
`deployment_status`	`str`	Status: 'active', 'pending', 'error'
`deployment_type`	`str`	Type: 'config_based', 'adk', 'file'
`endpoint_url`	`str \| None`	Agent invocation endpoint
`cloud_provider`	`str \| None`	Cloud provider (e.g., 'gcp')
`region`	`str \| None`	Deployment region
`hipaa_compliant`	`bool`	HIPAA compliance flag
`hipaa_verified`	`bool`	HIPAA verification status
`created_at`	`datetime \| None`	Creation timestamp
`deployed_at`	`datetime \| None`	Deployment timestamp

class DeploymentResult

DeploymentResultPydantic Model

Result of a deployment operation with success status and metadata.

`success`	`bool`	Whether deployment succeeded
`agent`	`DeployedAgent \| None`	Deployed agent (if successful)
`message`	`str`	Result message
`error`	`str \| None`	Error message (if failed)
`deployment_duration_seconds`	`float \| None`	Time taken to deploy

Complete Examples

Basic Agent Deployment

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

client = AgentClient()

agent = client.deploy(AgentConfig(
    name="customer-support",
    display_name="Customer Support Bot",
    description="Handles customer inquiries",
    instruction="You are a helpful customer support agent.",
    model="claude-3.5-sonnet",
    input_schema=InputSchema(fields={"message": StringField(required=True)}),
    output_schema=OutputSchema(fields={"response": StringField()}),
    timeout=60,
    memory_mb=512,
    version="1.0.0"
))

print(f"Deployed: {agent.endpoint_url}")

With Complex Schemas

from hikigai.agentsdk import ArrayField, ObjectField, IntegerField

config = AgentConfig(
    name="diagnosis-agent",
    display_name="Diagnosis Assistant",
    description="Provides diagnosis suggestions",
    instruction="Analyze symptoms and suggest diagnoses...",
    input_schema=InputSchema(
        fields={
            "symptoms": ArrayField(items=StringField()),
            "patient_age": IntegerField(minimum=0, maximum=150),
            "medical_history": ObjectField(properties={
                "conditions": ArrayField(),
                "medications": ArrayField()
            })
        }
    ),
    output_schema=OutputSchema(
        fields={
            "diagnoses": ArrayField(items=ObjectField(properties={
                "name": StringField(),
                "confidence": IntegerField(minimum=0, maximum=100),
                "icd10_code": StringField()
            })),
            "recommendations": ArrayField(items=StringField())
        }
    ),
    timeout=120,
    memory_mb=1024,
    version="1.0.0",
    hipaa_compliant=True
)

agent = client.deploy(config)

Managing Agents

# List all agents
agents = client.list_agents()
for agent in agents:
    print(f"{agent.name}: {agent.deployment_status}")

# Get specific agent
agent = client.get_agent("customer-support")
print(agent.version)
print(agent.endpoint_url)

# Delete agent
client.delete_agent("customer-support")

Deployment Methods

There are two ways to deploy agents on Hikigai: using SDK methods or calling the API directly. Choose based on your use case.

SDK Methods (.deploy)

The recommended approach for most use cases. Simple, validated, and handles all the complexity for you.

from hikigai.agentsdk import AgentClient, AgentConfig

client = AgentClient(api_key="your-api-key", project_id="your-project-id")
agent = client.deploy(AgentConfig(
    name="medical-coder",
    instruction="Extract ICD-10 codes...",
    model="claude-3.5-sonnet"
))

✓ Use SDK Method When:

• You're deploying from Python
• You want automatic validation and error handling
• You prefer a simple, one-liner deployment

Multi-Cloud Deployment

deploy() accepts an explicit cloud target. Omit these and the platform uses its configured default.

agent = client.deploy(
    config,
    cloud_service="gcp-cloud-run",  # pick from the cloud catalog (GCP or AWS)
    region="us-central1",
    cloud_credential_id="cred-123",  # omit for platform-managed
)

# Discover targets via AppSDK catalog
from hikigai.appsdk import AppClient
catalog = AppClient(api_key=..., project_id=...).cloud.catalog(workload="agent")
for service in catalog.services:
    print(service.id, service.display_name)

Events & Webhooks

client.events covers webhook subscriptions and live event streaming for agent deploy/delete, jobs, invocations, and storage.

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

Error Handling

DeploymentError / HikigaiError

Raised when deployment or API requests fail

from hikigai.agentsdk import HikigaiError, DeploymentError

try:
    agent = client.deploy(config)
except DeploymentError as e:
    print(f"Deployment failed: {e}")
except HikigaiError as e:
    print(f"API error: {e}")

ValidationError / ConfigurationError

Raised when configuration validation fails or credentials are missing

from hikigai.agentsdk import ValidationError, ConfigurationError

try:
    config = AgentConfig(name="invalid name!")  # Invalid chars
except ValidationError as e:
    print(f"Validation error: {e}")

AuthenticationError

Raised when the API key is invalid or missing

from hikigai.agentsdk import AuthenticationError

try:
    client = AgentClient(api_key="invalid-key", project_id="proj_123")
    agent = client.list_agents()
except AuthenticationError as e:
    print(f"Authentication failed: {e}")

← Back to SDK Overview View AppSDK Docs →