The way to Construct Contract-First Agentic Resolution Programs with PydanticAI for Threat-Conscious, Coverage-Compliant Enterprise AI

!pip -q set up -U pydantic-ai pydantic openai nest_asyncio


import os
import time
import asyncio
import getpass
from dataclasses import dataclass
from typing import Listing, Literal


import nest_asyncio
nest_asyncio.apply()


from pydantic import BaseModel, Discipline, field_validator
from pydantic_ai import Agent
from pydantic_ai.fashions.openai import OpenAIChatModel
from pydantic_ai.suppliers.openai import OpenAIProvider


OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
   attempt:
       from google.colab import userdata
       OPENAI_API_KEY = userdata.get("OPENAI_API_KEY")
   besides Exception:
       OPENAI_API_KEY = None
if not OPENAI_API_KEY:
   OPENAI_API_KEY = getpass.getpass("Enter OPENAI_API_KEY: ").strip()

class RiskItem(BaseModel):
   threat: str = Discipline(..., min_length=8)
   severity: Literal["low", "medium", "high"]
   mitigation: str = Discipline(..., min_length=12)




class DecisionOutput(BaseModel):
   choice: Literal["approve", "approve_with_conditions", "reject"]
   confidence: float = Discipline(..., ge=0.0, le=1.0)
   rationale: str = Discipline(..., min_length=80)
   identified_risks: Listing[RiskItem] = Discipline(..., min_length=2)
   compliance_passed: bool
   situations: Listing[str] = Discipline(default_factory=record)
   next_steps: Listing[str] = Discipline(..., min_length=3)
   timestamp_unix: int = Discipline(default_factory=lambda: int(time.time()))


   @field_validator("confidence")
   @classmethod
   def confidence_vs_risk(cls, v, data):
       dangers = data.knowledge.get("identified_risks") or []
       if any(r.severity == "excessive" for r in dangers) and v > 0.70:
           elevate ValueError("confidence too excessive given high-severity dangers")
       return v


   @field_validator("choice")
   @classmethod
   def reject_if_non_compliant(cls, v, data):
       if data.knowledge.get("compliance_passed") is False and v != "reject":
           elevate ValueError("non-compliant selections have to be reject")
       return v


   @field_validator("situations")
   @classmethod
   def conditions_required_for_conditional_approval(cls, v, data):
       d = data.knowledge.get("choice")
       if d == "approve_with_conditions" and (not v or len(v) < 2):
           elevate ValueError("approve_with_conditions requires a minimum of 2 situations")
       if d == "approve" and v:
           elevate ValueError("approve should not embrace situations")
       return v

@dataclass
class DecisionContext:
   company_policy: str
   risk_threshold: float = 0.6




mannequin = OpenAIChatModel(
   "gpt-5",
   supplier=OpenAIProvider(api_key=OPENAI_API_KEY),
)


agent = Agent(
   mannequin=mannequin,
   deps_type=DecisionContext,
   output_type=DecisionOutput,
   system_prompt="""
You're a company choice evaluation agent.
It's essential to consider threat, compliance, and uncertainty.
All outputs should strictly fulfill the DecisionOutput schema.
"""
)

@agent.output_validator
def ensure_risk_quality(consequence: DecisionOutput) -> DecisionOutput:
   if len(consequence.identified_risks) < 2:
       elevate ValueError("minimal two dangers required")
   if not any(r.severity in ("medium", "excessive") for r in consequence.identified_risks):
       elevate ValueError("a minimum of one medium or excessive threat required")
   return consequence




@agent.output_validator
def enforce_policy_controls(consequence: DecisionOutput) -> DecisionOutput:
   coverage = CURRENT_DEPS.company_policy.decrease()
   textual content = (
       consequence.rationale
       + " ".be part of(consequence.next_steps)
       + " ".be part of(consequence.situations)
   ).decrease()
   if consequence.compliance_passed:
       if not any(ok in textual content for ok in ["encryption", "audit", "logging", "access control", "key management"]):
           elevate ValueError("lacking concrete safety controls")
   return consequence

async def run_decision():
   world CURRENT_DEPS
   CURRENT_DEPS = DecisionContext(
       company_policy=(
           "No deployment of techniques dealing with private knowledge or transaction metadata "
           "with out encryption, audit logging, and least-privilege entry management."
       )
   )


   immediate = """
Resolution request:
Deploy an AI-powered buyer analytics dashboard utilizing a third-party cloud vendor.
The system processes consumer habits and transaction metadata.
Audit logging will not be applied and customer-managed keys are unsure.
"""


   consequence = await agent.run(immediate, deps=CURRENT_DEPS)
   return consequence.output




choice = asyncio.run(run_decision())


from pprint import pprint
pprint(choice.model_dump())