Complete reference for all Ainz-OS classes and methods.
register_agent(agent_id, agent)Register an agent with the kernel
kernel.register_agent("json_1", agent)enqueue_task(task)Add a task to the queue
kernel.enqueue_task(Task(id="t1", required_capability="validate"))
tick()Run one scheduling + execution cycle
status = kernel.tick()
# {"scheduled": 3, "executed": 2}get_status()Get current kernel status
status = kernel.get_status()
# {"total_agents": 5, "completed_tasks": 100, "failed_tasks": 2}Agent(name, capacity, capabilities)Create a new agent
agent = Agent(
name="json_processor",
capacity=5,
capabilities=["parse", "validate"]
)agent.stateCurrent agent state (IDLE, BUSY, FAILED)
if agent.state == AgentState.IDLE:
# agent is availableagent.active_task_countNumber of currently running tasks
print(f"Running: {agent.active_task_count}")Task(id, name, required_capability, priority, payload)Create a new task
task = Task(
id="task_001",
name="validate_json",
required_capability="validate",
priority=10,
payload={"data": user_json}
)scheduler.pending_count()Get number of pending tasks
while kernel.scheduler.pending_count() > 0:
kernel.tick()scheduler.schedule()Get next task-agent assignment
match = kernel.scheduler.schedule() # (task, agent) or None
monitor.check_health(agents)Get aggregate health metrics
metrics = kernel.monitor.check_health(agents)
print(f"Active: {metrics.active_agents}")
print(f"Avg util: {metrics.avg_utilization}%")monitor.get_agent_health(agent)Get per-agent health info
health = kernel.monitor.get_agent_health(agent)
# {
# "id": "agent_1",
# "state": "busy",
# "utilization": "60.0%",
# "is_healthy": true
# }