This repository has been archived on 2026-04-16. You can view files and clone it, but cannot push or open issues or pull requests.
llm-principal-broker/llm_broker/registrar_factory.py
Tyler King f3ffeb38ae feat: AgentRegistrar abstraction + Entra Agent ID driver
Extracts agent identity registration behind AgentRegistrar protocol.
Three implementations:
  KeycloakRegistrar — Keycloak Admin REST API (existing, refactored)
  EntraRegistrar    — Microsoft Entra Agent ID platform (NEW)
  StubRegistrar     — dev mode without real IdP

Driver selection via AGENT_REGISTRAR env var:
  auto     — prefers Entra if configured, Keycloak fallback, stub default
  keycloak — explicit Keycloak
  entra    — explicit Entra Agent ID

Entra integration:
  Registers agent as Entra app + service principal via Graph API
  Tags with delegation metadata (agent_type, delegator, governed:true)
  Client secret TTL matches delegation expiry
  Deletes application on revocation (cascades to SP)
  Uses msal for token acquisition
  Future: native Agent ID Blueprint API when GA

Files:
  registrar.py         — AgentRegistrar protocol + AgentCredentials dataclass
  registrar_keycloak.py — refactored from keycloak.py
  registrar_entra.py   — NEW Entra Graph API driver
  registrar_stub.py    — dev mode stub
  registrar_factory.py — driver selection factory
  delegation.py        — updated to use registrar abstraction
  settings.py          — added Entra config + agent_registrar field

All 7 smoke tests pass with stub registrar.
Implements GCAP-SPEC-LLM-PRINCIPAL-BROKER-0001 §4.1-§4.3.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-04 16:52:45 -04:00

64 lines
2.4 KiB
Python

"""Registrar factory — selects the appropriate AgentRegistrar based on config."""
import logging
from .registrar import AgentRegistrar
from .registrar_stub import StubRegistrar
logger = logging.getLogger(__name__)
def create_registrar(config) -> AgentRegistrar:
"""Create the appropriate registrar based on AGENT_REGISTRAR setting."""
driver = config.agent_registrar
if driver == "keycloak":
if not config.keycloak_admin_client_secret:
logger.warning("Keycloak secret not configured, using stub")
return StubRegistrar()
from .registrar_keycloak import KeycloakRegistrar
return KeycloakRegistrar(
base_url=config.keycloak_url,
realm=config.keycloak_realm,
client_id=config.keycloak_admin_client_id,
client_secret=config.keycloak_admin_client_secret,
)
elif driver == "entra":
if not config.entra_client_secret:
logger.warning("Entra secret not configured, using stub")
return StubRegistrar()
from .registrar_entra import EntraRegistrar
return EntraRegistrar(
tenant_id=config.entra_tenant_id,
client_id=config.entra_client_id,
client_secret=config.entra_client_secret,
agent_blueprint_id=config.entra_agent_blueprint_id,
)
elif driver == "auto":
if config.entra_client_secret:
from .registrar_entra import EntraRegistrar
logger.info("Auto-selected Entra registrar")
return EntraRegistrar(
tenant_id=config.entra_tenant_id,
client_id=config.entra_client_id,
client_secret=config.entra_client_secret,
agent_blueprint_id=config.entra_agent_blueprint_id,
)
elif config.keycloak_admin_client_secret:
from .registrar_keycloak import KeycloakRegistrar
logger.info("Auto-selected Keycloak registrar")
return KeycloakRegistrar(
base_url=config.keycloak_url,
realm=config.keycloak_realm,
client_id=config.keycloak_admin_client_id,
client_secret=config.keycloak_admin_client_secret,
)
else:
logger.warning("No IdP configured, using stub registrar")
return StubRegistrar()
else:
logger.warning("Unknown registrar driver: %s, using stub", driver)
return StubRegistrar()