Files
tudo-para-ia-mais-humana-pl…/tools/generate_mcp_control_contracts.py

629 lines
26 KiB
Python

"""Generate the MCP control contract registry for Mais Humana.
The generated module is source code on purpose. It is the compact,
importable registry that proves which platform/profile/surface/report contracts
Mais Humana can expose through the MCP control plane without reading Markdown
at runtime.
"""
from __future__ import annotations
from pathlib import Path
from typing import Iterable
ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
OUTPUT = SRC / "mais_humana" / "generated_mcp_control_contracts.py"
def ensure_import_path() -> None:
import sys
src = str(SRC)
if src not in sys.path:
sys.path.insert(0, src)
def q(value: object) -> str:
return repr(str(value))
def tuple_literal(values: Iterable[object], *, indent: int = 8) -> str:
cleaned = [str(value) for value in values if str(value).strip()]
if not cleaned:
return "()"
if len(cleaned) <= 30:
return "(" + ", ".join(q(value) for value in cleaned) + ("," if len(cleaned) == 1 else "") + ")"
pad = " " * indent
lines = ["("]
for value in cleaned:
lines.append(f"{pad}{q(value)},")
lines.append(" " * (indent - 4) + ")")
return "\n".join(lines)
def normalize(value: str) -> str:
return value.replace("_", "-").replace(" ", "-").replace("/", "-").lower()
def category_values(categories: Iterable[object]) -> tuple[str, ...]:
return tuple(getattr(category, "value", str(category)) for category in categories)
def audience_for_profile(profile_id: str, categories: tuple[str, ...]) -> str:
if profile_id in {"administrador_empresa"}:
return "ADMINISTRATOR"
if profile_id in {"ceo", "gestor_comercial"}:
return "EXECUTIVE"
if profile_id in {"gestor_operacional", "secretaria"}:
return "OPERATOR"
if profile_id in {"suporte", "atendimento_cliente"}:
return "SUPPORT"
if profile_id in {"cliente_externo"}:
return "CUSTOMER"
if profile_id in {"tecnico"}:
return "TECHNICAL"
if profile_id in {"juridico"} or "legal" in categories:
return "LEGAL"
if profile_id in {"financeiro", "contador"} or "finance" in categories:
return "FINANCE"
return "USER"
def truth_for_platform(platform_id: str) -> str:
if platform_id == "docs":
return "FORMAL_EXCEPTION"
if platform_id == "intelligence":
return "BLOCKED"
if platform_id in {"integracoes", "mcps", "identity", "business", "ui"}:
return "SAME_SOURCE_READY"
if platform_id in {"finance", "compliance", "public", "gettys", "customer_ops", "stj"}:
return "RESPONSE_READY"
return "DERIVED"
def panel_ready_for_platform(platform_id: str) -> bool:
return platform_id not in {"intelligence"}
def maturity_for_platform(platform_id: str) -> int:
if platform_id in {"mcps", "business", "identity", "ui"}:
return 9
if platform_id == "docs":
return 7
if platform_id == "intelligence":
return 3
if platform_id in {"integracoes", "finance", "compliance", "public"}:
return 8
return 6
def payload_fields(platform_id: str, profile_id: str, surface: str, category: str) -> tuple[str, ...]:
return (
"origin",
"destination",
"tool",
"payload",
"actor",
"permission",
"result",
"traceId",
"auditId",
"timestamp",
"projectId",
"platformId",
"profileId",
"surfaceId",
"category",
"sourceEndpoint",
"sourceToolId",
"sourcePayloadHash",
"sourceRecordsHash",
"truthState",
"panelReady",
"gptExplainable",
"humanNextAction",
f"{platform_id}Status",
f"{profile_id}Need",
f"{normalize(surface)}State",
f"{category}Gate",
)
def redaction_requirements(platform_id: str) -> tuple[str, ...]:
return (
"bloquear valores cfat_ brutos em qualquer artefato humano",
"permitir apenas credentialRef, tokenRef ou secretRef opacos",
"mascarar Authorization Bearer antes de persistir evidencia",
"registrar sourcePayloadHash e sourceRecordsHash sem payload sensivel bruto",
f"vincular redaction ao provider {platform_id} via MCP",
)
ACCESS_POLICY_SURFACES = (
("gpt-execute-probe", "governance", "mais_humana.gateway.access_policy.gpt_probe"),
("admin-ui-render", "experience", "mais_humana.gateway.access_policy.admin_ui"),
("automation-smoke", "observability", "mais_humana.gateway.access_policy.smoke"),
)
def unique_tuple(values: Iterable[object]) -> tuple[str, ...]:
seen: set[str] = set()
output: list[str] = []
for value in values:
text = str(value)
if not text or text in seen:
continue
seen.add(text)
output.append(text)
return tuple(output)
def access_policy_payload_fields(platform_id: str, profile_id: str, surface: str, category: str) -> tuple[str, ...]:
return unique_tuple(
payload_fields(platform_id, profile_id, surface, category)
+ (
"httpMethod",
"contentType",
"userAgent",
"authorizationCredentialRef",
"authorizationRawPersisted",
"wafDecision",
"wafRuleId",
"rateLimitPerMinute",
"logRetentionDays",
"requestHash",
"responseHash",
"redactionPolicyId",
"secretSafe",
"pluginCloudflareDiagnosticIgnored",
"wranglerOperationalReference",
)
)
def validation_steps(platform_id: str, profile_id: str, surface: str, kind: str) -> tuple[str, ...]:
return (
f"chamar {platform_id} somente via tudo-para-ia-mcps-internos-plataform",
f"validar envelope MCP completo para {profile_id}",
f"comparar payload GPT e UI para {normalize(surface)}",
"confirmar sourcePayloadHash e sourceRecordsHash",
"confirmar que o resultado nao contem segredo bruto",
f"registrar evidencia do contrato {kind}",
)
def access_policy_validations(platform_id: str, profile_id: str, surface: str) -> tuple[str, ...]:
return (
f"executar {surface} de {platform_id}/{profile_id} por /v1/execute com POST application/json",
"confirmar User-Agent operacional e separar WAF de erro runtime",
"confirmar Authorization via credentialRef sem persistir bearer bruto",
"validar traceId, auditId, requestHash e responseHash",
"validar redaction contra cfat_, bearer bruto e tokens numericos longos",
"registrar rate limit, retencao de logs e decisao WAF no MCP",
)
def contract_block(
name: str,
*,
contract_id: str,
kind: str,
platform_id: str,
profile_id: str,
tool_id: str,
title: str,
purpose: str,
source_tool_id: str,
payload: tuple[str, ...],
truth: str,
panel_ready: bool,
gpt_explainable: bool,
report_model_id: str,
audience: str,
redaction: tuple[str, ...],
validations: tuple[str, ...],
pending: str,
order_ids: tuple[str, ...],
policy_tags: tuple[str, ...],
maturity: int,
generated_from: str,
) -> str:
return "\n".join(
[
f"{name} = McpControlContract(",
f" contract_id={q(contract_id)},",
f" kind=McpContractKind.{kind},",
f" platform_id={q(platform_id)},",
f" profile_id={q(profile_id)},",
f" tool_id={q(tool_id)},",
f" title={q(title)},",
f" purpose={q(purpose)},",
" source_endpoint=MCP_EXECUTE_ENDPOINT,",
f" source_tool_id={q(source_tool_id)},",
" required_transit_fields=MCP_TRANSIT_FIELDS,",
f" required_payload_fields={tuple_literal(payload)},",
f" truth_state=TruthState.{truth},",
f" panel_ready={panel_ready},",
f" gpt_explainable={gpt_explainable},",
f" report_model_id={q(report_model_id)},",
f" audience=AudienceClass.{audience},",
f" redaction_requirements={tuple_literal(redaction)},",
f" validation_steps={tuple_literal(validations)},",
f" pending_if_missing={q(pending)},",
f" order_ids={tuple_literal(order_ids)},",
f" policy_tags={tuple_literal(policy_tags)},",
f" maturity_level={int(maturity)},",
f" generated_from={q(generated_from)},",
")",
]
)
def build_contracts() -> tuple[list[str], list[str]]:
ensure_import_path()
from mais_humana.catalog import HUMAN_PROFILES, PLATFORMS
blocks: list[str] = []
names: list[str] = []
index = 1
for platform in PLATFORMS:
platform_categories = category_values(platform.primary_categories)
surfaces = tuple(platform.expected_surfaces or ("readiness", "evidence", "admin_ui"))
first_surface = surfaces[0]
first_category = platform_categories[0] if platform_categories else "governance"
truth = truth_for_platform(platform.platform_id)
panel_ready = panel_ready_for_platform(platform.platform_id)
maturity = maturity_for_platform(platform.platform_id)
for profile in HUMAN_PROFILES:
profile_categories = category_values(profile.priority_needs)
common = tuple(value for value in platform_categories if value in profile_categories)
category = common[0] if common else first_category
audience = audience_for_profile(profile.profile_id, profile_categories)
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=f"{platform.platform_id}.{profile.profile_id}.provider-tool",
kind="PROVIDER_TOOL",
platform_id=platform.platform_id,
profile_id=profile.profile_id,
tool_id=f"mais_humana.{platform.platform_id}.{profile.profile_id}.explain",
title=f"{platform.title} explicado para {profile.name}",
purpose=(
f"Permitir que {profile.name} receba leitura humana de {platform.title} "
"por provider MCP minimo da Mais Humana."
),
source_tool_id="mais_humana.rulebook.compact",
payload=payload_fields(platform.platform_id, profile.profile_id, first_surface, category),
truth=truth,
panel_ready=panel_ready,
gpt_explainable=True,
report_model_id=f"report.{platform.platform_id}.{profile.profile_id}.operacional",
audience=audience,
redaction=redaction_requirements(platform.platform_id),
validations=validation_steps(platform.platform_id, profile.profile_id, first_surface, "provider_tool"),
pending=f"publicar e validar tool MCP para {platform.platform_id}/{profile.profile_id}",
order_ids=("0016_EXECUTIVA__publicar-rulebook-mcp-provider-mais-humana", "0029_GERENCIAL__pactuar-mcp-como-caminho-unico-administrativo"),
policy_tags=("mcp_provider", "same_source", "human_explanation", category),
maturity=maturity,
generated_from="platform_profile_provider_contract",
)
)
index += 1
for profile in HUMAN_PROFILES:
profile_categories = category_values(profile.priority_needs)
category = platform_categories[0] if platform_categories else "governance"
audience = audience_for_profile(profile.profile_id, profile_categories)
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=f"{platform.platform_id}.{profile.profile_id}.report-model",
kind="REPORT_MODEL",
platform_id=platform.platform_id,
profile_id=profile.profile_id,
tool_id="mais_humana.relatorios.modelos_oficiais",
title=f"Modelo oficial de relatorio {platform.title} para {profile.name}",
purpose=(
f"Definir modelo oficial que separa artefato de apoio de relatorio oficial para {profile.name}."
),
source_tool_id="mais_humana.relatorios.modelos_oficiais",
payload=payload_fields(platform.platform_id, profile.profile_id, "relatorio-oficial", category),
truth=truth,
panel_ready=panel_ready,
gpt_explainable=True,
report_model_id=f"report.{platform.platform_id}.{profile.profile_id}.oficial",
audience=audience,
redaction=redaction_requirements(platform.platform_id),
validations=validation_steps(platform.platform_id, profile.profile_id, "relatorio-oficial", "report_model"),
pending=f"homologar modelo oficial de relatorio para {platform.platform_id}/{profile.profile_id}",
order_ids=("0030_GERENCIAL__definir-modelo-oficial-de-relatorio-humano",),
policy_tags=("report_model", "official_report", "audience", audience.lower()),
maturity=maturity,
generated_from="platform_profile_report_model_contract",
)
)
index += 1
for surface in surfaces:
surface_id = normalize(surface)
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=f"{platform.platform_id}.{surface_id}.ui-screen",
kind="UI_SCREEN",
platform_id=platform.platform_id,
profile_id="todos",
tool_id="mais_humana.admin_ui.same_source",
title=f"Tela {surface} de {platform.title} renderizada pela UI",
purpose=(
f"Impedir fonte paralela na tela {surface}; GPT e painel devem usar o mesmo contrato MCP."
),
source_tool_id="mais_humana.admin_ui.same_source",
payload=payload_fields(platform.platform_id, "todos", surface, first_category),
truth=truth,
panel_ready=panel_ready,
gpt_explainable=True,
report_model_id=f"screen.{platform.platform_id}.{surface_id}.same-source",
audience="ADMINISTRATOR",
redaction=redaction_requirements(platform.platform_id),
validations=validation_steps(platform.platform_id, "todos", surface, "ui_screen"),
pending=f"validar same-source da tela {surface} em {platform.platform_id}",
order_ids=("0020_EXECUTIVA__validar-mcp-admin-ui-same-source-rulebook", "0031_GERENCIAL__governar-ui-como-renderizador-sem-fonte-paralela"),
policy_tags=("ui_renderer", "screen", "same_source", surface_id),
maturity=maturity,
generated_from="platform_surface_ui_contract",
)
)
index += 1
for operation in ("consulta", "diagnostico", "acao", "auditoria", "explicacao"):
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=f"{platform.platform_id}.{operation}.transit-policy",
kind="TRANSIT_POLICY",
platform_id=platform.platform_id,
profile_id="todos",
tool_id=f"mais_humana.transito.{operation}",
title=f"Politica de transito {operation} para {platform.title}",
purpose=(
f"Garantir que {operation} de {platform.title} registre origem, destino, tool, ator, permissao, resultado, trace, audit e timestamp."
),
source_tool_id=f"mais_humana.transito.{operation}",
payload=payload_fields(platform.platform_id, "todos", operation, first_category),
truth=truth,
panel_ready=panel_ready,
gpt_explainable=True,
report_model_id=f"policy.{platform.platform_id}.{operation}.mcp-transit",
audience="ADMINISTRATOR",
redaction=redaction_requirements(platform.platform_id),
validations=validation_steps(platform.platform_id, "todos", operation, "transit_policy"),
pending=f"validar politica MCP de {operation} para {platform.platform_id}",
order_ids=("0029_GERENCIAL__pactuar-mcp-como-caminho-unico-administrativo",),
policy_tags=("mcp_transit", operation, "audit", "trace"),
maturity=maturity,
generated_from="platform_transit_policy_contract",
)
)
index += 1
for ref_kind in ("credentialRef", "tokenRef", "secretRef", "cfat"):
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=f"{platform.platform_id}.{normalize(ref_kind)}.redaction-policy",
kind="REDACTION_POLICY",
platform_id=platform.platform_id,
profile_id="todos",
tool_id="mais_humana.redaction.guard",
title=f"Redaction {ref_kind} para {platform.title}",
purpose=(
f"Provar que artefatos humanos de {platform.title} nao propagam valor sensivel bruto de {ref_kind}."
),
source_tool_id="mais_humana.redaction.guard",
payload=payload_fields(platform.platform_id, "todos", ref_kind, "security"),
truth=truth,
panel_ready=True,
gpt_explainable=True,
report_model_id=f"redaction.{platform.platform_id}.{normalize(ref_kind)}",
audience="TECHNICAL",
redaction=redaction_requirements(platform.platform_id) + (f"validar {ref_kind} como referencia opaca",),
validations=validation_steps(platform.platform_id, "todos", ref_kind, "redaction_policy"),
pending=f"executar varredura redaction para {platform.platform_id}/{ref_kind}",
order_ids=("0019_EXECUTIVA__blindar-redaction-credentialref-e-cfat-em-artefatos",),
policy_tags=("redaction", "secret_policy", normalize(ref_kind)),
maturity=max(6, maturity),
generated_from="platform_redaction_policy_contract",
)
)
index += 1
for profile in HUMAN_PROFILES:
profile_categories = category_values(profile.priority_needs)
audience = audience_for_profile(profile.profile_id, profile_categories)
for surface, category, tool_id in ACCESS_POLICY_SURFACES:
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=f"{platform.platform_id}.{profile.profile_id}.{surface}.access-policy",
kind="ACCESS_POLICY",
platform_id=platform.platform_id,
profile_id=profile.profile_id,
tool_id=tool_id,
title=f"Politica de acesso {surface} para {platform.title} e {profile.name}",
purpose=(
f"Garantir que chamadas GPT/MCP de {platform.title} para {profile.name} "
"usem headers, bearer redigido, WAF classificado, hashes e ledger auditavel."
),
source_tool_id="mais_humana.gateway.access_policy",
payload=access_policy_payload_fields(platform.platform_id, profile.profile_id, surface, category),
truth=truth,
panel_ready=panel_ready,
gpt_explainable=True,
report_model_id=f"access.{platform.platform_id}.{profile.profile_id}.{normalize(surface)}",
audience=audience,
redaction=redaction_requirements(platform.platform_id)
+ (
"bloquear persistencia de Authorization Bearer bruto",
"registrar apenas credentialRef e hashes de evidencia",
),
validations=access_policy_validations(platform.platform_id, profile.profile_id, surface),
pending=f"homologar politica de acesso {surface} para {platform.platform_id}/{profile.profile_id}",
order_ids=(
"0032_EXECUTIVA__validar-live-tools-mais-humana-v1-execute-com-evidencia",
"0045_GERENCIAL__pactuar-politica-acesso-waf-gpt-mcp-gateway",
),
policy_tags=("access_policy", "waf", "redaction", "same_source", normalize(surface)),
maturity=max(7, maturity),
generated_from="platform_profile_access_policy_contract",
)
)
index += 1
special_contracts = [
(
"docs.formal-exception.docs-catalogonly",
"DOCS_EXCEPTION",
"docs",
"todos",
"mais_humana.docs.formal_exception",
"Excecao formal Docs catalogOnly",
"Formalizar que Docs pode permanecer catalogOnly quando a decisao institucional assim declarar.",
"mais_humana.docs.formal_exception",
"docs-catalogonly",
"documentation",
"FORMAL_EXCEPTION",
True,
True,
"policy.docs.catalogonly.formal-exception",
"LEGAL",
("0017_EXECUTIVA__resolver-docs-catalogonly-com-response-ready-minimo",),
("docs_exception", "catalogonly", "governance"),
7,
),
(
"mais-humana.canonical-rename.platform",
"CANONICAL_RENAME",
"mais_humana",
"todos",
"mais_humana.canonical.rename",
"Nome canonico administrativo tudo-para-ia-mais-humana-platform",
"Registrar o alias canonico administrativo sem executar migracao local/remota destrutiva.",
"mais_humana.canonical.rename",
"canonical-rename",
"governance",
"DOCUMENTED",
True,
True,
"policy.mais_humana.rename.platform",
"ADMINISTRATOR",
("0033_GERENCIAL__autorizar-renome-canonico-mais-humana-plataform",),
("canonical_identity", "rename", "external_decision"),
6,
),
]
for special in special_contracts:
(
contract_id,
kind,
platform_id,
profile_id,
tool_id,
title,
purpose,
source_tool_id,
surface,
category,
truth,
panel_ready,
gpt_explainable,
report_model_id,
audience,
order_ids,
tags,
maturity,
) = special
name = f"CONTRACT_{index:04d}"
names.append(name)
blocks.append(
contract_block(
name,
contract_id=contract_id,
kind=kind,
platform_id=platform_id,
profile_id=profile_id,
tool_id=tool_id,
title=title,
purpose=purpose,
source_tool_id=source_tool_id,
payload=payload_fields(platform_id, profile_id, surface, category),
truth=truth,
panel_ready=panel_ready,
gpt_explainable=gpt_explainable,
report_model_id=report_model_id,
audience=audience,
redaction=redaction_requirements(platform_id),
validations=validation_steps(platform_id, profile_id, surface, kind.lower()),
pending=f"registrar decisao institucional para {contract_id}",
order_ids=order_ids,
policy_tags=tags,
maturity=maturity,
generated_from="special_governance_contract",
)
)
index += 1
return blocks, names
def render_module(blocks: list[str], names: list[str]) -> str:
header = '''"""Generated MCP control contracts for Mais Humana.
Do not edit this file by hand. Regenerate it with:
python tools/generate_mcp_control_contracts.py
The contracts are Python source so they can be imported, tested, packaged,
hashed, and exposed as a provider payload without parsing external YAML.
"""
from __future__ import annotations
from .human_rulebook import MCP_TRANSIT_FIELDS, TruthState
from .mcp_contract import AudienceClass, MCP_EXECUTE_ENDPOINT, McpContractKind, McpControlContract
'''
assignments = "\n\n".join(blocks)
contract_lines = ["CONTRACTS = ("]
for name in names:
contract_lines.append(f" {name},")
contract_lines.append(")")
all_lines = ["__all__ = [", " 'CONTRACTS',", "]"]
return header + assignments + "\n\n" + "\n".join(contract_lines) + "\n\n" + "\n".join(all_lines) + "\n"
def main() -> int:
blocks, names = build_contracts()
OUTPUT.write_text(render_module(blocks, names), encoding="utf-8")
print(f"generated {len(names)} MCP contracts at {OUTPUT}")
return 0
if __name__ == "__main__":
raise SystemExit(main())