Skip to content

Identity

Agent identity registry and clearance enforcement.

identity

Agent identity normalization and enforcement helpers.

AgentIdentityRegistry

Registry for declared agent identities and permission scopes.

Source code in safeai/core/identity.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class AgentIdentityRegistry:
    """Registry for declared agent identities and permission scopes."""

    def __init__(self, identities: list[AgentIdentity] | None = None) -> None:
        self._identities: dict[str, AgentIdentity] = {}
        self.load(identities or [])

    def load(self, identities: list[AgentIdentity]) -> None:
        self._identities = {item.agent_id: item for item in identities}

    def get(self, agent_id: str) -> AgentIdentity | None:
        return self._identities.get(str(agent_id).strip())

    def has(self, agent_id: str) -> bool:
        return self.get(agent_id) is not None

    def all(self) -> list[AgentIdentity]:
        return list(self._identities.values())

    def validate(
        self,
        *,
        agent_id: str,
        tool_name: str | None = None,
        data_tags: list[str] | None = None,
    ) -> AgentIdentityValidationResult:
        token = str(agent_id).strip()
        tags = list(data_tags or [])
        if not token:
            return AgentIdentityValidationResult(
                allowed=False,
                reason="agent identity is required",
                unauthorized_tags=sorted({str(tag).strip().lower() for tag in tags if str(tag).strip()}),
                identity=None,
            )

        if not self._identities:
            return AgentIdentityValidationResult(
                allowed=True,
                reason="agent identity registry is not configured",
                unauthorized_tags=[],
                identity=None,
            )

        identity = self.get(token)
        if identity is None:
            return AgentIdentityValidationResult(
                allowed=False,
                reason=f"agent '{token}' is not declared",
                unauthorized_tags=sorted({str(tag).strip().lower() for tag in tags if str(tag).strip()}),
                identity=None,
            )

        if tool_name and identity.tools and str(tool_name).strip() not in identity.tools:
            return AgentIdentityValidationResult(
                allowed=False,
                reason=f"agent '{token}' is not bound to tool '{tool_name}'",
                unauthorized_tags=[],
                identity=identity,
            )

        unauthorized = _find_unauthorized_tags(
            tags=tags,
            clearance_tags=identity.clearance_tags,
        )
        if unauthorized:
            return AgentIdentityValidationResult(
                allowed=False,
                reason=f"agent '{token}' exceeds tag clearance: {','.join(unauthorized)}",
                unauthorized_tags=unauthorized,
                identity=identity,
            )

        return AgentIdentityValidationResult(
            allowed=True,
            reason="agent identity allows tool and data scope",
            unauthorized_tags=[],
            identity=identity,
        )

normalize_agent_identities 

normalize_agent_identities(raw_items: list[dict[str, Any]]) -> list[AgentIdentity]

Normalize YAML/JSON identity documents into runtime objects.

Source code in safeai/core/identity.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def normalize_agent_identities(raw_items: list[dict[str, Any]]) -> list[AgentIdentity]:
    """Normalize YAML/JSON identity documents into runtime objects."""
    identities: list[AgentIdentity] = []
    seen_ids: set[str] = set()

    for item in raw_items:
        doc = AgentIdentityDocumentModel.model_validate(item)
        models: list[AgentIdentityModel] = []
        if doc.agent is not None:
            models.append(doc.agent)
        if doc.agents:
            models.extend(doc.agents)

        for model in models:
            name = model.agent_id.strip()
            if name in seen_ids:
                raise ValueError(f"Duplicate agent identity: {name}")
            seen_ids.add(name)
            identities.append(
                AgentIdentity(
                    agent_id=name,
                    description=model.description,
                    tools=set(model.tools),
                    clearance_tags={tag.lower() for tag in model.clearance_tags},
                )
            )
    return identities