"""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 "RESPONSE_READY" 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 8 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"), ) ADMINISTRATION_OPERATIONS = ( ("consulta", "query", "mcp.admin.readonly", "mais_humana.admin.consulta"), ("diagnostico", "diagnostic", "mcp.admin.diagnostic", "mais_humana.admin.diagnostico"), ("acao", "action", "mcp.admin.action.request", "mais_humana.admin.acao"), ("auditoria", "audit", "mcp.admin.audit", "mais_humana.admin.auditoria"), ("explicacao", "explanation", "mcp.admin.explain", "mais_humana.admin.explicacao"), ) 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 administration_route_payload_fields( platform_id: str, profile_id: str, operation: str, route_family: str, permission: str, category: str, ) -> tuple[str, ...]: return unique_tuple( payload_fields(platform_id, profile_id, operation, category) + ( "adminRouteId", "adminRouteKind", "controlPlaneId", "ownerPlatformId", "targetPlatformId", "routeFamily", "permissionScope", "executionMode", "capabilityId", "capabilityVersion", "policyDecision", "policyReason", "inputSchemaHash", "outputSchemaHash", "operatorIntent", "approvalRequired", "dryRunSupported", "rollbackSupported", "humanExplanation", "auditRetentionPolicy", "mcpOnlyAdministration", "directPlatformBypassBlocked", f"{operation}Route", f"{route_family}Family", permission.replace(".", "_"), ) ) 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 administration_route_validations(platform_id: str, profile_id: str, operation: str, permission: str) -> tuple[str, ...]: return ( f"executar rota administrativa {operation} para {platform_id}/{profile_id} exclusivamente pelo MCPs Internos", "confirmar origin, destination, tool, payload, actor, permission, result, traceId, auditId e timestamp", f"validar permissionScope={permission} e bloquear bypass direto da plataforma", "registrar inputSchemaHash, outputSchemaHash, sourcePayloadHash e sourceRecordsHash", "confirmar que acao mutavel exige approvalRequired ou dryRunSupported conforme contrato", "gerar humanExplanation para auditoria da IA administradora", ) 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 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) for operation, route_family, permission, tool_id in ADMINISTRATION_OPERATIONS: route_id = f"{platform.platform_id}.{profile.profile_id}.{operation}.administration-route" name = f"CONTRACT_{index:04d}" names.append(name) blocks.append( contract_block( name, contract_id=route_id, kind="ADMINISTRATION_ROUTE", platform_id=platform.platform_id, profile_id=profile.profile_id, tool_id=tool_id, title=f"Rota administrativa {operation} de {platform.title} para {profile.name}", purpose=( f"Expor {operation} de {platform.title} para {profile.name} somente pelo MCP control plane, " "com decisao de permissao, auditoria, explicacao humana e bloqueio de bypass direto." ), source_tool_id="mais_humana.mcp_transit.ledger", payload=administration_route_payload_fields( platform.platform_id, profile.profile_id, operation, route_family, permission, category, ), truth=truth, panel_ready=panel_ready, gpt_explainable=True, report_model_id=f"admin-route.{platform.platform_id}.{profile.profile_id}.{operation}", audience=audience, redaction=redaction_requirements(platform.platform_id) + ( "bloquear administracao direta fora do MCPs Internos", "persistir apenas hashes e referencias opacas de credencial", ), validations=administration_route_validations(platform.platform_id, profile.profile_id, operation, permission), pending=f"homologar rota administrativa {operation} para {platform.platform_id}/{profile.profile_id}", order_ids=( "0029_GERENCIAL__pactuar-mcp-como-caminho-unico-administrativo", "0044_GERENCIAL__institucionalizar-ledger-transito-mcp-como-gate-release", "0048_GERENCIAL__homologar-politica-acesso-gpt-mcp-como-gate-institucional", ), policy_tags=("administration_route", "mcp_only", route_family, operation, permission), maturity=max(7, maturity), generated_from="platform_profile_administration_route_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 transitoria Docs catalogOnly", ( "Garantir que surface Docs sem responseReady vire pendencia rastreavel; catalogOnly e apenas " "excecao transitoria com prazo, owner, evidencia MCP e plano de promocao operacional." ), "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 a decisao institucional aprovada e manter aliases de compatibilidade sem executar migracao local/remota destrutiva.", "mais_humana.canonical.rename", "canonical-rename", "governance", "SAME_SOURCE_READY", True, True, "policy.mais_humana.rename.platform", "ADMINISTRATOR", ( "0035_EXECUTIVA__reconciliar-nome-canonico-real-alias-platform", "0047_GERENCIAL__decidir-nome-canonico-e-politica-alias-mais-humana", ), ("canonical_identity", "rename", "approved", "alias_compatibility"), 8, ), ] 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"manter canonico aprovado e migrar fisicamente apenas em janela segura 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())