Skip to content

Contracts

Tool contract registry and validation.

contracts

Tool contract normalization and request validation helpers.

ToolContractRegistry

Runtime registry for declared tool contracts.

Source code in safeai/core/contracts.py 
 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
class ToolContractRegistry:
    """Runtime registry for declared tool contracts."""

    def __init__(self, contracts: list[ToolContract] | None = None) -> None:
        self._contracts: dict[str, ToolContract] = {}
        self.load(contracts or [])

    def load(self, contracts: list[ToolContract]) -> None:
        self._contracts = {item.tool_name: item for item in contracts}

    def get(self, tool_name: str) -> ToolContract | None:
        return self._contracts.get(str(tool_name).strip())

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

    def validate_request(self, tool_name: str, data_tags: list[str]) -> ContractValidationResult:
        contract = self.get(tool_name)
        if contract is None:
            return ContractValidationResult(
                allowed=False,
                reason=f"tool '{tool_name}' has no declared contract",
                unauthorized_tags=sorted(set(data_tags)),
                contract=None,
            )

        if not data_tags:
            return ContractValidationResult(
                allowed=True,
                reason="no classified data tags on request",
                unauthorized_tags=[],
                contract=contract,
            )

        unauthorized: list[str] = []
        accepted = {tag.lower() for tag in contract.accepts_tags}
        for raw_tag in data_tags:
            token = str(raw_tag).strip().lower()
            if not token:
                continue
            expanded = expand_tag_hierarchy([token])
            if accepted.intersection(expanded):
                continue
            unauthorized.append(token)

        if unauthorized:
            return ContractValidationResult(
                allowed=False,
                reason=f"tool '{tool_name}' does not accept data tags: {','.join(sorted(set(unauthorized)))}",
                unauthorized_tags=sorted(set(unauthorized)),
                contract=contract,
            )

        return ContractValidationResult(
            allowed=True,
            reason="tool contract allows request tags",
            unauthorized_tags=[],
            contract=contract,
        )

    def all(self) -> list[ToolContract]:
        return list(self._contracts.values())

normalize_contracts 

normalize_contracts(raw_items: list[dict[str, Any]]) -> list[ToolContract]

Normalize YAML/JSON contract documents into runtime contract objects.

Source code in safeai/core/contracts.py 
106
107
108
109
110
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
138
139
140
141
def normalize_contracts(raw_items: list[dict[str, Any]]) -> list[ToolContract]:
    """Normalize YAML/JSON contract documents into runtime contract objects."""
    contracts: list[ToolContract] = []
    seen_names: set[str] = set()

    for item in raw_items:
        doc = ToolContractDocumentModel.model_validate(item)
        models: list[ToolContractModel] = []
        if doc.contract is not None:
            models.append(doc.contract)
        if doc.contracts:
            models.extend(doc.contracts)

        for model in models:
            name = model.tool_name.strip()
            if name in seen_names:
                raise ValueError(f"Duplicate tool contract name: {name}")
            seen_names.add(name)
            contracts.append(
                ToolContract(
                    tool_name=name,
                    description=model.description,
                    accepts_tags={tag.lower() for tag in model.accepts.tags},
                    accepts_fields=set(model.accepts.fields),
                    emits_tags={tag.lower() for tag in model.emits.tags},
                    emits_fields=set(model.emits.fields),
                    stores_fields=set(model.stores.fields),
                    stores_retention=model.stores.retention,
                    side_effects=ToolSideEffects(
                        reversible=model.side_effects.reversible,
                        requires_approval=model.side_effects.requires_approval,
                        description=model.side_effects.description,
                    ),
                )
            )
    return contracts