How you talk to the Crew gateway depends on where your code runs. This page maps each surface to the right API.
| Where your code runs | Use |
|---|---|
| Dashboard UI page (TypeScript / React) | @kirocrew/app-sdk hooks (resolved by the host at runtime) |
| Python app, CLI tool, or service | kirocrew-client package — pip install kirocrew-client |
| Node.js / Electron app | Call Gateway REST/WebSocket endpoints directly via fetch() |
There is no published TypeScript gateway-client npm package. The kirocrew-client method names below describe the canonical Gateway API surface — the same endpoints any client (including raw fetch) talks to.
Dashboard UI pages import permission-scoped hooks from @kirocrew/app-sdk, resolved at runtime via the host import map:
import { useAppApi, useAppEvents } from '@kirocrew/app-sdk' function MyPage() { const api = useAppApi() // permission-scoped GET/POST/PUT/PATCH/DELETE useAppEvents('notification', (e) => console.log(e)) // ... }
useAppApi() returns a client whose methods (get, post, put, patch, del) call the Gateway endpoints listed below, scoped to your permissions.api allowlist. Out-of-scope paths throw an error. The host injects auth automatically.
| Hook | Returns | Purpose |
|---|---|---|
useAppApi() | AppApiClient | Permission-scoped HTTP client |
useAppEvents(event, cb) | () => void | Subscribe to WebSocket events; returns unsubscribe |
useTheme() | Theme | Reactive theme (mode, accent, colorTheme) |
useAppInfo() | AppInfo | App metadata (name, version, permissions) |
useNavigate() | (path) => void | Navigate to Crew routes |
useNotify() | (text, opts?) => void | Show a toast notification |
useNavBadge() | (count) => void | Update sidebar badge count |
useChatLauncher() | LauncherFn | Navigate to chat with optional agent + message |
Import from @kirocrew/app-sdk/ui:
Card, CardTitle, Btn, SendBtn, Input, SearchInput, Badge, AimBadge, StatCard, Skeleton, ContentSkeleton, EmptyState, PageHeader, Toggle, InfoTip, SegmentedControl, MarkdownRenderer.
pip install kirocrew-client
Standalone async client using aiohttp. No dependency on the Crew main package.
from kirocrew_client import CrewClient async with CrewClient(app_name="my-app") as mc: ok = await mc.ping() slots = await mc.list_slots() task_id = await mc.dispatch_agent_async("my-agent", "Analyze ticket T-123") result = await mc.get_task_result(task_id)
CrewClient( base_url="", # default: http://localhost:{KIROCREW_PORT or 5476} token="", # optional for localhost app_name="", # enables app-scoped storage + auto-auth timeout=30, # request timeout seconds max_retries=3, retry_base_delay=1.0, message_length_limit=40000, on_auth_expired=None, # async callback returning new token )
When app_name is set and no explicit auth is provided, the client auto-reads the app secret from ~/.kiro/crew/apps/{name}/.app_secret and exchanges it for a short-lived token via POST /api/apps/{name}/token.
Method names below use snake_case for Python; TypeScript names shown for readability. The same endpoints are reachable via raw fetch() from Node or a browser.
| TS name | Python | Purpose |
|---|---|---|
authenticate() | authenticate() | Exchange app secret for token (auto-called if appName set) |
setToken(token) | set_token(token) | Manually set auth token on HTTP and WebSocket |
| TS name | Python | Purpose |
|---|---|---|
ping() | ping() | Check gateway reachability |
getStatus() | get_status() | Gateway health (version, uptime, slots, provider) |
getSystemInfo() | get_system_info() | CPU, memory, disk metrics |
| TS name | Python | Purpose |
|---|---|---|
createSlot(name, agent?) | create_slot(name, agent="") | Create a new chat session |
listSlots() | list_slots() | List all active sessions |
deleteSlot(id) | delete_slot(id) | Remove a session |
getSlotHistory(id, limit?) | get_slot_history(id, limit=50) | Get slot message history |
sendMessage(id, msg) | send_message(id, msg) | Send a message (validates length, auto-flushes pending context) |
| TS name | Python | Purpose |
|---|---|---|
connect() | (auto) | Open WebSocket connection |
disconnect() | (auto) | Close WebSocket connection |
onChatChunk(slotId, cb) | subscribe | Stream response chunks for a slot |
onChatDone(slotId, cb) | subscribe | Response complete for a slot |
onNotification(cb) | subscribe | Receive notifications |
onToolCall(cb) | subscribe | Receive tool call events |
onConnectionChange(cb) | subscribe | Connection state changes |
onRaw(cb) | subscribe | All parsed WebSocket events |
All subscription methods return an unsubscribe function.
Event types (partial list): chat_chunk, chat_done, chat_message, chat_error, tool_call, notification, slots, slot_title, dashboard, log, refresh, approval, subagent_done, task_update, task_complete, proactive_notification, app_reload, error.
| TS name | Python | Purpose |
|---|---|---|
spawn(task, agent?) | spawn(task, agent="") | Spawn a background subagent |
spawnMany(tasks, agents?) | spawn_many(tasks, agents=None) | Spawn multiple subagents in parallel |
listSubagents() | list_subagents() | List all subagents |
getSubagentStatus(id) | get_subagent_status(id) | Get subagent output |
| TS name | Python | Purpose |
|---|---|---|
addCron(name, opts) | add_cron(name, **opts) | Create a scheduled job |
listCrons() | list_crons() | List all cron jobs |
updateCron(id, opts) | update_cron(id, **opts) | Update a cron job |
removeCron(id) | remove_cron(id) | Delete a cron job |
pauseCron(id) | pause_cron(id) | Pause without deleting |
resumeCron(id) | resume_cron(id) | Resume a paused job |
| TS name | Python | Purpose |
|---|---|---|
addLesson(rule, cat, scope?) | add_lesson(rule, cat, scope="") | Save a learned rule |
listLessons() | list_lessons() | List all lessons |
removeLesson(query) | remove_lesson(query) | Remove matching lessons |
| TS name | Python | Purpose |
|---|---|---|
sendNotification(text, opts?) | send_notification(text, **opts) | Send via Slack or dashboard |
listNotifications() | list_notifications() | List notifications |
ackNotifications() | ack_notifications() | Acknowledge all |
| TS name | Python | Purpose |
|---|---|---|
approveAction(slot, task) | approve_action(slot, task) | Approve a pending tool action |
rejectAction(slot, task) | reject_action(slot, task) | Reject a pending tool action |
resolveApproval(id, ok) | resolve_approval(id, ok) | Resolve an approval by ID |
getApprovalMode() | get_approval_mode() | Get current approval mode |
setApprovalMode(mode) | set_approval_mode(mode) | Set to "auto" or "interactive" |
| TS name | Python | Purpose |
|---|---|---|
listModels() | list_models() | List available LLM models |
setSlotModel(slotId, model) | set_slot_model(slot, model) | Set model for a slot |
| TS name | Python | Purpose |
|---|---|---|
listMcpServers() | list_mcp_servers() | List registered MCP servers |
registerMcpServer(def) | register_mcp_server(name, cmd, args?, env?) | Register an MCP server |
removeMcpServer(name) | remove_mcp_server(name) | Remove an MCP server |
registerAppMcp(name, entry) | register_app_mcp(name, ...) | Write MCP entry to ~/.kiro/crew/mcp.json |
unregisterAppMcp(name) | unregister_app_mcp(name) | Remove MCP entry |
| TS name | Python | Purpose |
|---|---|---|
dispatchAgent(agent, prompt) | dispatch_agent(agent, prompt) | Run agent synchronously |
dispatchAgentAsync(agent, prompt) | dispatch_agent_async(agent, prompt) | Run in background |
getTaskResult(taskId) | get_task_result(id) | Poll task status |
| TS name | Purpose |
|---|---|
installAgentConfig(name, config) | Install agent JSON to ~/.kiro/agents/ (merges mcpServers) |
removeAgentConfig(name) | Remove agent config |
installSkill(name, srcDir) | Copy skill directory to ~/.kiro/crew/skills/ |
removeSkill(name) | Remove skill directory |
| TS name | Python | Purpose |
|---|---|---|
getGatewayConfig(key) | get_gateway_config(key) | Read gateway config section |
setGatewayConfig(key, value) | set_gateway_config(key, val) | Write gateway config section |
| TS name | Python | Purpose |
|---|---|---|
getAppDataDir() | get_app_data_dir() → Path | App-scoped data directory |
getAppConfig() | get_app_config() | Read app config via REST |
setAppConfig(config) | set_app_config(cfg) | Write app config via REST |
| TS name | Python | Purpose |
|---|---|---|
memorySearch(query, topK?) | memory_search(q, top_k=8) | Semantic memory search |
Silent background context — appears in the next user-initiated turn without triggering a response.
| TS name | Python | Purpose |
|---|---|---|
injectContext(slotId, content, opts?) | inject_context(slot, content, ...) | Inject context (null slotId = buffer locally) |
flushPendingContext(slotId) | flush_pending_context(slot) | Flush buffered entries |
setDefaultSlot(slotId) | set_default_slot(slot) | Auto-flush on sendMessage |
Options: { source?: string, ephemeral?: boolean, maxAge?: number }.
For app backends: verify a request was signed by the gateway reverse proxy.
| Function | Language | Purpose |
|---|---|---|
verifyProxyRequest(req, appName, opts?) | Node.js | Verify HMAC on any Node.js request |
verify_proxy_request(request, app_name, ...) | Python | Verify HMAC on any aiohttp / Django / FastAPI request |
verify_proxy_request_raw(header, ...) | Python | Verify from a raw header string |
Signature is HMAC-SHA256(timestamp:method:/api/path[?query]:sha256(body)) with the app secret as the key. Timestamps must be within ±60 s of now. Uses constant-time comparison.
For app-lifecycle operations beyond the client wrappers:
| Method | Path | Purpose |
|---|---|---|
| GET | /api/apps | List all installed apps |
| GET | /api/apps/registry | List available apps from registry |
| GET | /api/apps/blob?repo=&path=&ref= | Proxy images from a registry app's git repo |
| POST | /api/apps/install | Install from local path |
| POST | /api/apps/register | Register a self-managed app |
| POST | /api/apps/registry/install | Install from registry |
| GET | /api/apps/{name} | Get app details |
| GET | /api/apps/{name}/manifest | Get app manifest |
| GET / PUT | /api/apps/{name}/config | Read / write app config |
| POST | /api/apps/{name}/update | Update installed app |
| POST | /api/apps/{name}/uninstall | Uninstall app |
| POST | /api/apps/{name}/enable | Enable app |
| POST | /api/apps/{name}/disable | Disable app |
| POST | /api/apps/{name}/dev | Toggle dev mode (body {"enabled": bool}) |
| POST | /api/apps/{name}/open | Launch app via openCommand |
| GET | /apps/{name}/ui/{path} | Serve app UI bundle files |
| * | /apps/{name}/api/{path} | Reverse proxy to app backend (HMAC-signed) |
kirocrew-client also ships two helpers for install automation.
Validate and serialize app.json:
from kirocrew_client import AppManifest m = AppManifest.from_dict({"name": "my-app", "version": "1.0.0", ...}) errors = m.validate() # list[str] — empty if valid data = m.to_dict()
Manage app installation via the Gateway REST API:
from kirocrew_client import CrewClient, AppLifecycle async with CrewClient() as mc: lifecycle = AppLifecycle(mc) await lifecycle.install("/path/to/my-app") await lifecycle.enable("my-app") await lifecycle.disable("my-app") await lifecycle.uninstall("my-app") apps = await lifecycle.list()
Manage the Crew gateway process (start, stop, health check):
from kirocrew_client import GatewayManager gm = GatewayManager(port=5476) await gm.start() healthy = await gm.is_healthy() await gm.stop()
All kirocrew-client errors are CrewError instances with code, message, status, body.
| Code | Trigger | Retried? |
|---|---|---|
AUTH_REQUIRED | Remote connection without token | No |
AUTH_EXPIRED | 401 / 403 response | No (calls on_auth_expired if set) |
VALIDATION_ERROR | Invalid input | No |
NOT_FOUND | 404 response | No |
RATE_LIMITED | 429 response | Yes (Retry-After or backoff) |
SERVER_ERROR | 5xx response | Yes (exponential backoff) |
NETWORK_ERROR | Timeout or connection failure | Yes (exponential backoff) |
WS_DISCONNECTED | WebSocket not connected | No |
from kirocrew_client import CrewError try: await mc.send_message("slot-1", "hello") except CrewError as e: print(e.code, e.message, e.status)
SDK / API reference