Agents API Reference

agents.registry

Classes

AgentsRegistry

Registro de agentes — instanciable, no singleton.

La instancia global legacy 'agents_registry' se mantiene para backward compat. Para nuevos entry points o tests, crear AgentsRegistry() directamente.

Source code in agents/registry.py

class AgentsRegistry:
    """Registro de agentes — instanciable, no singleton.

    La instancia global legacy 'agents_registry' se mantiene para backward compat.
    Para nuevos entry points o tests, crear AgentsRegistry() directamente.
    """

    def __init__(self):
        self._global_agents: dict[str, Callable] = {}
        self._global_profiles: dict[str, dict] = {}
        self._workspace_agents: dict[str, Callable] = {}
        self._workspace_profiles: dict[str, dict] = {}

    # ---------- Registro global ----------
    def register_global(self, agent_type: str, profile: dict | None = None):
        def decorator(func: Callable) -> Callable:
            normalized = agent_type.lower()
            if normalized in self._global_agents:
                logger.warning(f"Agent global '{normalized}' ya registrado – sobrescribiendo.")
            self._global_agents[normalized] = func
            if profile:
                self._global_profiles[normalized] = profile
            logger.info(f"Agent global registrado: {normalized}")
            return func

        return decorator

    # ---------- Registro de workspace ----------
    def register_workspace_agent(
        self, agent_type: str, func: Callable, profile: dict | None = None
    ):
        normalized = agent_type.lower()
        self._workspace_agents[normalized] = func
        if profile:
            self._workspace_profiles[normalized] = profile
        logger.info(f"Agent workspace registrado: {normalized}")

    def clear_workspace_agents(self):
        self._workspace_agents.clear()
        self._workspace_profiles.clear()
        logger.info("Agentes del workspace descargados")

    # ---------- Consulta ----------
    def get_agent(self, agent_type: str) -> Callable | None:
        normalized = agent_type.lower()
        if normalized in self._workspace_agents:
            return self._workspace_agents[normalized]
        return self._global_agents.get(normalized)

    def get_profile(self, agent_type: str) -> dict | None:
        normalized = agent_type.lower()
        if normalized in self._workspace_profiles:
            return self._workspace_profiles[normalized]
        return self._global_profiles.get(normalized)

    def list_agents(self) -> dict[str, Callable]:
        return {**self._global_agents, **self._workspace_agents}

    def list_global_agents(self) -> dict[str, Callable]:
        return self._global_agents.copy()

    def clear(self):
        self._global_agents.clear()
        self._global_profiles.clear()
        self._workspace_agents.clear()
        self._workspace_profiles.clear()

agents.loader

Functions

load_workspace_agents

load_workspace_agents(workspace_name: str)

Carga los agentes definidos en workspaces//agents/*.yaml

Source code in agents/loader.py

def load_workspace_agents(workspace_name: str):
    """Carga los agentes definidos en workspaces/<workspace>/agents/*.yaml"""
    from core.path_resolver import paths

    agents_dir = paths.workspace_agents_dir(workspace_name)
    if not agents_dir.exists():
        logger.info(f"No hay agentes locales en el workspace '{workspace_name}'.")
        return

    for agent_file in agents_dir.glob("*.yaml"):
        if agent_file.name.startswith("_"):
            continue  # plantillas (p.ej. _FULL_TEMPLATE.yaml) no son agentes
        try:
            with open(agent_file, encoding="utf-8") as f:
                profile = yaml.safe_load(f)
            if not profile or "name" not in profile:
                logger.warning(f"Archivo {agent_file} inválido, falta 'name'")
                continue

            name = profile["name"]

            # Solution: bind name value at definition time
            async def agent_func(
                task: str,
                history: list,
                pdf_text: str = "",
                tools_output: str = "",
                _name: str = name,
            ):
                return await _execute_specialized_agent(
                    _name, task, history, pdf_text, tools_output
                )

            agents_registry.register_workspace_agent(name, agent_func, profile)
            logger.info(f"Agente '{name}' cargado desde {agent_file}")
        except Exception as e:
            logger.error(f"Error cargando agente desde {agent_file}: {e}")

agents.profiles

Global built-in agent profiles. Only 'conversacional' is kept as the universal minimum fallback agent. All other agents are defined per workspace in workspaces//agents/*.yaml. Initial templates live in templates/agents/ for new workspaces.

agents.base

Ejecuta cualquier agente especializado definido en los perfiles (globales o del workspace). La configuración (prompt, longitud, temperatura, clave de memoria, etc.) se toma del perfil registrado en AgentsRegistry.

agents.service

Classes

AgentsService

Servicio centralizado para ejecutar agentes especializados.

Source code in agents/service.py

class AgentsService:
    """Servicio centralizado para ejecutar agentes especializados."""

    @staticmethod
    def get_available_agents():
        """Devuelve los 6 agentes especializados (nombres exactos)."""
        from agents.registry import agents_registry

        return list(agents_registry.list_agents().keys())

    @staticmethod
    async def execute_agent(agent_type: str, query: str, history: list, on_stream_chunk=None):
        """Ejecuta cualquier agente registrado."""
        try:
            from agents.registry import agents_registry

            agent_func = agents_registry.get_agent(agent_type)
            if not agent_func:
                return f"❌ Agent '{agent_type}' no encontrado."

            if on_stream_chunk:
                from agents.base import _execute_specialized_agent

                return await _execute_specialized_agent(
                    agent_type, query, history, on_stream_chunk=on_stream_chunk
                )
            return await agent_func(query, history)
        except Exception as e:
            logger.error(f"Error ejecutando agent {agent_type}: {e}")
            return f"❌ Error en el agente {agent_type}: {str(e)[:200]}"

    @staticmethod
    async def categorize_task(task: str) -> str:
        """Classify task using available agents from the registry."""
        try:
            from agents.registry import agents_registry

            available = list(agents_registry.list_agents().keys())
            agent_list = ", ".join(available)

            prompt = (
                "Classify this task into one exact word from this list: "
                f"{agent_list}. Reply ONLY with the word in lowercase.\n\n"
                f"Task: {task}"
            )

            response = await models.call(
                messages=[{"role": "user", "content": prompt}],
                role="fast",
                temperature=0.0,
                max_tokens=20,
            )

            category = response.choices[0].message.content.strip().lower()
            return category if category in available else settings.fallback_agent

        except Exception:
            logger.warning("Fallback in categorize_task")
            return settings.fallback_agent

Functions

get_available_agents staticmethod

get_available_agents()

Devuelve los 6 agentes especializados (nombres exactos).

Source code in agents/service.py

@staticmethod
def get_available_agents():
    """Devuelve los 6 agentes especializados (nombres exactos)."""
    from agents.registry import agents_registry

    return list(agents_registry.list_agents().keys())

execute_agent async staticmethod

execute_agent(
    agent_type: str,
    query: str,
    history: list,
    on_stream_chunk=None,
)

Ejecuta cualquier agente registrado.

Source code in agents/service.py

@staticmethod
async def execute_agent(agent_type: str, query: str, history: list, on_stream_chunk=None):
    """Ejecuta cualquier agente registrado."""
    try:
        from agents.registry import agents_registry

        agent_func = agents_registry.get_agent(agent_type)
        if not agent_func:
            return f"❌ Agent '{agent_type}' no encontrado."

        if on_stream_chunk:
            from agents.base import _execute_specialized_agent

            return await _execute_specialized_agent(
                agent_type, query, history, on_stream_chunk=on_stream_chunk
            )
        return await agent_func(query, history)
    except Exception as e:
        logger.error(f"Error ejecutando agent {agent_type}: {e}")
        return f"❌ Error en el agente {agent_type}: {str(e)[:200]}"

categorize_task async staticmethod

categorize_task(task: str) -> str

Classify task using available agents from the registry.

Source code in agents/service.py

@staticmethod
async def categorize_task(task: str) -> str:
    """Classify task using available agents from the registry."""
    try:
        from agents.registry import agents_registry

        available = list(agents_registry.list_agents().keys())
        agent_list = ", ".join(available)

        prompt = (
            "Classify this task into one exact word from this list: "
            f"{agent_list}. Reply ONLY with the word in lowercase.\n\n"
            f"Task: {task}"
        )

        response = await models.call(
            messages=[{"role": "user", "content": prompt}],
            role="fast",
            temperature=0.0,
            max_tokens=20,
        )

        category = response.choices[0].message.content.strip().lower()
        return category if category in available else settings.fallback_agent

    except Exception:
        logger.warning("Fallback in categorize_task")
        return settings.fallback_agent

agents.audit

Audit Log — registro de operaciones sensibles.

Registra: bash commands, file deletions, git force pushes. Almacenamiento en archivo JSON lines para simplicidad.

Functions

log_operation

log_operation(
    operation: str,
    details: str = "",
    user: str = "morphix",
    success: bool = True,
) -> None

Registra una operación en el audit log.

Source code in agents/audit.py

def log_operation(
    operation: str,
    details: str = "",
    user: str = "morphix",
    success: bool = True,
) -> None:
    """Registra una operación en el audit log."""
    _ensure_audit_dir()
    entry = {
        "timestamp": datetime.now(UTC).isoformat(),
        "operation": operation,
        "details": details[:500],
        "user": user,
        "success": success,
    }
    try:
        with open(AUDIT_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception as e:
        logger.warning(f"No se pudo escribir audit log: {e}")

get_recent_operations

get_recent_operations(limit: int = 50) -> list[dict]

Lee las últimas N operaciones del audit log.

Source code in agents/audit.py

def get_recent_operations(limit: int = 50) -> list[dict]:
    """Lee las últimas N operaciones del audit log."""
    _ensure_audit_dir()
    if not AUDIT_FILE.exists():
        return []
    entries = []
    try:
        with open(AUDIT_FILE, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    try:
                        entries.append(json.loads(line))
                    except json.JSONDecodeError:
                        continue
    except Exception:
        return []
    return entries[-limit:]