Files
agent_contas/contas_mcp/servers/contas_mcp_server/line_policy.py

52 lines
1.8 KiB
Python

from __future__ import annotations
from dataclasses import dataclass
from typing import Any
POLICY_NAME = "authenticated_line_only"
POLICY_DESCRIPTION = "Somente a linha identificada/autenticada na chamada pode ser consultada ou alterada."
def _digits(value: Any) -> str:
return "".join(ch for ch in str(value or "") if ch.isdigit())
@dataclass(frozen=True)
class LinePolicyDecision:
allowed: bool
effective_msisdn: str
reason: str = ""
user_message: str = ""
requested_reference: dict[str, Any] | None = None
def apply_line_policy(tool_name: str, args: dict[str, Any]) -> LinePolicyDecision:
authenticated = _digits(args.get("msisdn"))
reference = args.get("requested_line_reference")
if not authenticated or not isinstance(reference, dict):
return LinePolicyDecision(True, authenticated or str(args.get("msisdn") or ""))
kind = str(reference.get("kind") or "").strip().lower()
requested = _digits(reference.get("value"))
same_line = False
if kind == "full" and requested:
same_line = requested == authenticated or requested.endswith(authenticated) or authenticated.endswith(requested)
elif kind == "suffix" and requested:
same_line = authenticated.endswith(requested)
else:
return LinePolicyDecision(True, authenticated)
if same_line:
return LinePolicyDecision(True, authenticated, requested_reference=reference)
return LinePolicyDecision(
False,
authenticated,
reason="other_line_not_allowed",
user_message=(
"Por segurança, este atendimento só permite consultar ou realizar operações "
"na linha identificada na chamada. Não posso usar outra linha informada na conversa."
),
requested_reference=reference,
)