Pydantic AI

pip install veriproof-sdk veriproof-sdk-instrumentation-pydantic-ai pydantic-ai

# .env
VERIPROOF_API_KEY=vp_live_...
VERIPROOF_APPLICATION_ID=my-ai-app
ANTHROPIC_API_KEY=sk-ant-...

import asyncio
import os
from veriproof_sdk import (
    configure_veriproof,
    VeriproofClientOptions,
    DecisionType,
    RiskLevel,
    SessionIntent,
    SessionOutcome,
    StepOutcome,
    StepType,
)
from veriproof_sdk_instrumentation_pydantic_ai import VeriproofPydanticAISession
from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel
 
 
# 1. Configure VeriProof once at application startup
configure_veriproof(
    VeriproofClientOptions(
        api_key=os.environ["VERIPROOF_API_KEY"],
        application_id=os.environ["VERIPROOF_APPLICATION_ID"],
    ),
    service_name=os.environ["VERIPROOF_APPLICATION_ID"],
    set_global=True,
)
 
 
async def assess_loan_application(applicant_id: str, credit_score: int) -> dict:
    # 2. Create the Pydantic AI agent
    agent = Agent(
        AnthropicModel("claude-3-5-haiku-20241022"),
        system_prompt=(
            "You are a credit risk analyst. "
            "Assess the creditworthiness of the applicant and recommend APPROVE or DECLINE."
        ),
    )
 
    # 3. Create a VeriProof governance session (async context manager)
    async with VeriproofPydanticAISession(
        application_id=os.environ["VERIPROOF_APPLICATION_ID"],
        session_name="loan-risk-assessment",
        intent=SessionIntent.DECISION_SUPPORT,
    ) as session:
 
        # 4. Add a step before the agent runs
        await session.add_step(
            step_name="creditworthiness-assessment",
            step_type=StepType.ANALYSIS,
            input={"applicant_id": applicant_id, "credit_score": credit_score},
        )
 
        # 5. Run the agent — model spans captured automatically
        result = await session.run(
            agent,
            f"Applicant {applicant_id} has a credit score of {credit_score}. "
            "Debt-to-income ratio is 0.32. Assess their loan application.",
        )
 
        # 6. Record the decision
        recommendation = "APPROVE" if credit_score >= 680 else "DECLINE"
        await session.set_decision(
            decision_type=DecisionType.APPROVAL if recommendation == "APPROVE" else DecisionType.REJECTION,
            decision_summary=result.data,
            risk_level=RiskLevel.LOW if credit_score >= 720 else RiskLevel.MEDIUM,
            outcome=StepOutcome.COMPLETED,
        )
 
        session.set_outcome(SessionOutcome.SUCCESS)
 
    return {
        "session_id": session.session_id,
        "merkle_root": session.merkle_root,
        "recommendation": recommendation,
    }
 
 
if __name__ == "__main__":
    result = asyncio.run(assess_loan_application("A-1042", credit_score=712))
    print(f"Session ID  : {result['session_id']}")
    print(f"Merkle root : {result['merkle_root']}")
    print(f"Decision    : {result['recommendation']}")

from pydantic_ai import Agent, RunContext, tool
 
agent = Agent(
    AnthropicModel("claude-3-5-sonnet-20241022"),
    system_prompt="You are a financial data assistant.",
)
 
@agent.tool
async def get_credit_report(ctx: RunContext, applicant_id: str) -> dict:
    # This tool call appears as a child span under the agent span
    return {"score": 712, "bureau": "Equifax", "inquiries": 2}