QAE-Claude-mcp-example

Instructions

Implementation Reference

• src/qae_mcp_server/server.py:107-227 (handler)

  The 'certify_action' function, decorated with @server.tool(), acts as the primary handler for certifying actions. It validates inputs, calls the 'SafetyCertifier', and constructs a response containing the certification results.

  def certify_action(
      action_id: str,
      agent_id: str,
      scope: float,
      reversibility: float,
      sensitivity: float,
      description: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Evaluate the safety of an action across multiple constraint dimensions.
  
      This tool uses the QAE safety kernel to assess whether a proposed action
      is safe to execute. It returns a certificate with a decision (Certified,
      CertifiedWithWarning, EscalateToHuman, or Blocked) and detailed margins
      for each constraint dimension.
  
      Args:
          action_id: Unique identifier for the action (e.g., "act_123")
          agent_id: Identifier of the agent proposing the action (e.g., "claude_v3")
          scope: Scope dimension [0, 1], where 0=narrow, 1=global
          reversibility: Reversibility dimension [0, 1], where 0=permanent, 1=easily reversible
          sensitivity: Sensitivity dimension [0, 1], where 0=low-impact, 1=high-impact
          description: Optional human-readable description of the action
  
      Returns:
          Dictionary with keys:
          - decision: "Certified", "CertifiedWithWarning", "EscalateToHuman", or "Blocked"
          - zone: "Safe", "Caution", or "Danger"
          - margins: Dict of dimension -> margin value [0, 1]
          - binding_constraint: Name of most restrictive constraint (if any)
          - drift_budget: Remaining budget after this certification
          - certificate_id: Unique certificate identifier
          - deterministic_hash: SHA256 hash of the certificate
          - timestamp: ISO 8601 timestamp
          - description: Echo of input description (if provided)
  
      Example:
          >>> certify_action(
          ...     action_id="act_deploy_algo",
          ...     agent_id="claude_sales",
          ...     scope=0.7,
          ...     reversibility=0.4,
          ...     sensitivity=0.8,
          ...     description="Deploy new recommendation algorithm to 10% of users"
          ... )
      """
      try:
          certifier = _get_certifier()
  
          # Validate input ranges
          for dim, val in [("scope", scope), ("reversibility", reversibility), ("sensitivity", sensitivity)]:
              if not (0.0 <= val <= 1.0):
                  raise ValueError(f"{dim} must be in [0, 1], got {val}")
  
          # Create state deltas from dimension scores
          # These represent changes to the agent's state in each dimension
          state_deltas = [
              StateDelta(dimension="scope_score", from_value=0.0, to_value=scope),
              StateDelta(dimension="reversibility_score", from_value=1.0, to_value=reversibility),
              StateDelta(dimension="sensitivity_score", from_value=0.0, to_value=sensitivity),
          ]
  
          # Create and certify the action
          action = SimpleAction(
              action_id=action_id,
              agent_id=agent_id,
              state_deltas=state_deltas,
          )
  
          certificate = certifier.certify(action)
  
          # Build response
          response = {
              "decision": certificate.decision,
              "zone": certificate.zone,
              "margins": certificate.margins,
              "binding_constraint": certificate.binding_constraint,
              "drift_budget": certificate.drift_budget,
              "certificate_id": certificate.certificate_id,
              "deterministic_hash": certificate.deterministic_hash,
              "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
          }
  
          if description:
              response["description"] = description
  
          # Log to certification history
          history_entry = {
              "certificate_id": certificate.certificate_id,
              "action_id": action_id,
              "agent_id": agent_id,
              "decision": certificate.decision,
              "zone": certificate.zone,
              "timestamp": response["timestamp"],
              "description": description,
          }
          _certification_history.append(history_entry)
          global _total_certifications
          _total_certifications += 1
  
          logger.info(
              f"Certified action {_sanitize_for_log(action_id)}: {certificate.decision} "
              f"(zone={certificate.zone}, cert_id={certificate.certificate_id})"
          )
  
          return response
  
      except ValueError as e:
          logger.error(f"Validation error in certify_action: {e}")
          return {
              "error": str(e),
              "action_id": action_id,
              "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
          }
      except Exception as e:
          logger.error(f"Error in certify_action: {e}")
          return {
              "error": f"Certification failed: {str(e)}",
              "action_id": action_id,
              "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
          }