Pydantic AI

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

# .env
VERIPROOF_API_KEY=vp_cust_your-tenant.<key-material>
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,
    AgentRole,
)
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,
)
 
# 2. Create the Pydantic AI agent once (reuse across requests)
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."
    ),
)
 
 
async def assess_loan_application(applicant_id: str, credit_score: int) -> dict:
    # 3. Configure governance attributes for this session
    decision = DecisionType.APPROVAL if credit_score >= 680 else DecisionType.DENIAL
    risk = RiskLevel.LOW if credit_score >= 720 else RiskLevel.MEDIUM
 
    session = VeriproofPydanticAISession(
        session_intent=SessionIntent.ADVISORY,
        decision_type=decision,
        agent_role=AgentRole.PRIMARY,
    )
    session.set_risk_level(risk.value)
 
    # 4. Run the agent — OTel spans are 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.",
    )
 
    recommendation = "APPROVE" if credit_score >= 680 else "DECLINE"
    return {
        "session_id": session.session_id,
        "recommendation": recommendation,
        "analysis": result.data,
    }
 
 
if __name__ == "__main__":
    result = asyncio.run(assess_loan_application("A-1042", credit_score=712))
    print(f"Session ID  : {result['session_id']}")
    print(f"Decision    : {result['recommendation']}")
    print(f"Analysis    : {result['analysis']}")

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}