99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
"""Public API facade for Agent OS runtime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
RUNTIME_DIR = Path(__file__).resolve().parent
|
|
if str(RUNTIME_DIR) not in sys.path:
|
|
sys.path.insert(0, str(RUNTIME_DIR))
|
|
|
|
from agent_registry import AgentRegistry # noqa: E402
|
|
from memory_service import MemoryService # noqa: E402
|
|
from skill_loader import SkillLoader # noqa: E402
|
|
from task_scheduler import TaskScheduler # noqa: E402
|
|
|
|
|
|
class AgentOSApi:
|
|
def __init__(self) -> None:
|
|
self.registry = AgentRegistry()
|
|
self.memory = MemoryService()
|
|
self.skills = SkillLoader()
|
|
self.scheduler = TaskScheduler(registry=self.registry)
|
|
|
|
def register_agent(
|
|
self,
|
|
agent_id: str,
|
|
name: str,
|
|
capabilities: list[str] | None = None,
|
|
skills: list[str] | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
return self.registry.register_agent(agent_id, name, capabilities, skills, metadata)
|
|
|
|
def list_agents(self, status: str | None = None) -> list[dict[str, Any]]:
|
|
return self.registry.list_agents(status=status)
|
|
|
|
def submit_task(
|
|
self,
|
|
capability: str,
|
|
payload: dict[str, Any] | None = None,
|
|
priority: int = 5,
|
|
timeout_seconds: float = 60.0,
|
|
max_retries: int = 1,
|
|
) -> dict[str, Any]:
|
|
return self.scheduler.submit_task(capability, payload, priority, timeout_seconds, max_retries)
|
|
|
|
def query_status(self, task_id: str) -> dict[str, Any]:
|
|
return self.scheduler.query_status(task_id)
|
|
|
|
def memory_store(
|
|
self,
|
|
key: str,
|
|
value: Any,
|
|
kind: str = "working",
|
|
scope: str = "default",
|
|
tags: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
return self.memory.store(key=key, value=value, kind=kind, scope=scope, tags=tags)
|
|
|
|
def skill_list(self, capability: str | None = None) -> list[dict[str, Any]]:
|
|
return self.skills.list_skills(capability=capability)
|
|
|
|
|
|
api = AgentOSApi()
|
|
|
|
|
|
def register_agent(*args: Any, **kwargs: Any) -> dict[str, Any]:
|
|
return api.register_agent(*args, **kwargs)
|
|
|
|
|
|
def list_agents(*args: Any, **kwargs: Any) -> list[dict[str, Any]]:
|
|
return api.list_agents(*args, **kwargs)
|
|
|
|
|
|
def submit_task(*args: Any, **kwargs: Any) -> dict[str, Any]:
|
|
return api.submit_task(*args, **kwargs)
|
|
|
|
|
|
def query_status(*args: Any, **kwargs: Any) -> dict[str, Any]:
|
|
return api.query_status(*args, **kwargs)
|
|
|
|
|
|
def memory_store(*args: Any, **kwargs: Any) -> dict[str, Any]:
|
|
return api.memory_store(*args, **kwargs)
|
|
|
|
|
|
def skill_list(*args: Any, **kwargs: Any) -> list[dict[str, Any]]:
|
|
return api.skill_list(*args, **kwargs)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
register_agent("agent.local", "Local Agent", capabilities=["self-test"], skills=["core"])
|
|
task = submit_task("self-test", {"message": "hello"})
|
|
memory_store("last_task", task["task_id"], kind="working")
|
|
print({"agents": list_agents(), "task": query_status(task["task_id"]), "skills": skill_list()})
|