Agent Evaluation

from rail_score_sdk import RailScoreClient
from rail_score_sdk.agent.exceptions import AgentBlockedError

client = RailScoreClient(api_key="YOUR_RAIL_API_KEY")

try:
    decision = client.agent.evaluate_tool_call(
        tool_name="credit_scoring",
        tool_params={
            "applicant_id": "usr_123",
            "zip_code": "10001",
            "annual_income": 65000,
        },
        domain="finance",
        mode="basic",           # "basic" | "deep"
        compliance_frameworks=["eu_ai_act", "gdpr"],
    )

    print(decision.decision)          # "ALLOW" | "FLAG" | "BLOCK"
    print(decision.decision_reason)   # human-readable reason
    print(decision.rail_score.score)  # overall RAIL score (0-10)
    print(decision.compliance_violations)   # list of detected violations
    print(decision.suggested_params)  # safer parameter alternatives (if any)
    print(decision.credits_consumed)  # float

    if decision.decision == "FLAG":
        # Proceed with caution — log and monitor
        result = run_tool(decision.suggested_params or tool_params)

except AgentBlockedError as e:
    print(f"Tool call blocked: {e.tool_name}")
    print(f"Reason: {e.decision_reason}")

result_data = run_tool(tool_params)

risk = client.agent.evaluate_tool_result(
    tool_name="web_search",
    tool_result_data=result_data,      # dict, str, or list
    tool_params=tool_params,           # original params for context
    mode="basic",
)

print(risk.pii_detected)              # bool
print(risk.pii_entities)              # list of detected PII types
print(risk.injection_detected)        # bool
print(risk.rail_score.score)          # overall RAIL score
print(risk.redacted_result)           # result with PII replaced by [TYPE]

check = client.agent.check_injection(
    content=user_input,
    content_source="user_message",   # "user_message" | "tool_result" | "external_data"
)

print(check.detected)           # bool
print(check.confidence)         # float 0-1
print(check.attack_type)        # "direct_instruction_override" | "role_hijack" | "jailbreak" | ...
print(check.severity)           # "critical" | "high" | "medium" | "low"
print(check.recommended_action) # "block" | "flag" | "allow"

if check.detected and check.severity in ("critical", "high"):
    raise ValueError(f"Injection attempt blocked: {check.attack_type}")

from rail_score_sdk.agent.exceptions import PlanBlockedError

plan_steps = [
    {"step": 1, "tool": "web_search",     "params": {"query": "user address lookup"}},
    {"step": 2, "tool": "send_email",     "params": {"to": "{search_result}", "body": "..."}},
    {"step": 3, "tool": "update_record",  "params": {"table": "users", "data": "..."}},
]

try:
    evaluation = client.agent.evaluate_plan(
        plan_steps=plan_steps,
        domain="general",
        mode="basic",
    )

    print(evaluation.overall_decision)    # "ALLOW" | "FLAG" | "BLOCK"
    print(evaluation.overall_risk_score)  # float 0-10
    for step in evaluation.step_evaluations:
        print(f"Step {step.step_number}: {step.decision} — {step.decision_reason}")

except PlanBlockedError as e:
    print(f"Plan blocked at step {e.blocked_step}: {e.reason}")

from rail_score_sdk.agent.session import AgentSession

session = AgentSession(
    client=client,
    agent_id="my-agent-001",
    max_session_age=3600,    # session expires after 1 hour
)

# Same API as client.agent — but risk is tracked cumulatively
decision = session.evaluate_tool_call(
    tool_name="credit_scoring",
    tool_params={"applicant_id": "usr_123", "income": 45000},
    domain="finance",
    mode="basic",
)

risk = session.evaluate_tool_result(
    tool_name="credit_scoring",
    tool_result_data=result,
    tool_params=tool_params,
)

# Review session-wide risk summary
summary = session.risk_summary()
print(summary.total_calls)        # int
print(summary.average_risk)       # "low" | "medium" | "high" | "critical"
print(summary.dimension_trends)   # dict of per-dimension avg scores
print(summary.patterns)           # list of detected cross-call patterns

session.close()

from rail_score_sdk.agent.policy import AgentPolicy, AgentPolicyEngine

engine = AgentPolicyEngine(
    policies={
        "credit_scoring": AgentPolicy(
            block_below=6.0,
            flag_below=8.0,
            dimension_thresholds={"fairness": 7.0, "privacy": 7.0},
        ),
        "send_email": AgentPolicy(
            block_below=5.0,
            flag_below=7.0,
        ),
    },
    default_policy=AgentPolicy(block_below=4.0, flag_below=6.0),
)

decision = client.agent.evaluate_tool_call(
    tool_name="credit_scoring",
    tool_params=params,
    domain="finance",
    mode="basic",
)

# Engine applies the credit_scoring policy to the decision
enforced = engine.evaluate("credit_scoring", decision)
print(enforced.final_action)   # "block" | "flag" | "allow"

from rail_score_sdk.agent.middleware import AgentMiddleware

middleware = AgentMiddleware(
    rail_api_key="YOUR_RAIL_API_KEY",
    policy="block",        # "block" | "log_only" | "suggest_fix" | "auto_fix"
    threshold=7.0,
    domain="finance",
    scan_results=True,     # Also scan tool output after execution
)

@middleware.guard(tool_name="credit_scoring")
async def run_credit_check(applicant_id, annual_income, zip_code):
    # This function is only called if evaluation returns ALLOW or FLAG
    return await credit_api.check(applicant_id, annual_income, zip_code)

# Call as normal — RAIL evaluation happens automatically
result = await run_credit_check(
    applicant_id="usr_123",
    annual_income=65000,
    zip_code="10001",
)

registry = client.agent.registry

# List all tools visible to your org (system defaults + your overrides)
tools = registry.list_tools()
for tool in tools:
    print(f"{tool['tool_name']}: risk={tool['risk_level']}, depth={tool['evaluation_depth']}")

# Register a custom tool profile
registry.register_tool({
    "tool_name": "loan_approval",
    "risk_level": "high",               # "low" | "medium" | "high" | "critical"
    "evaluation_depth": "deep",         # "basic" | "deep"
    "compliance_frameworks": ["eu_ai_act", "gdpr"],
    "proxy_variable_watchlist": ["zip_code", "neighborhood", "surname"],
    "prohibited_attributes": ["race", "religion", "national_origin"],
    "description": "Automated loan approval — requires deep evaluation",
})

# Remove a custom profile (reverts to system default)
registry.delete_tool("loan_approval")

from rail_score_sdk.agent.exceptions import (
    AgentBlockedError,
    PlanBlockedError,
    SessionClosedError,
)

try:
    decision = client.agent.evaluate_tool_call(
        tool_name="loan_approval",
        tool_params=params,
        domain="finance",
        mode="deep",
    )
except AgentBlockedError as e:
    print(f"Blocked: tool={e.tool_name}, reason={e.decision_reason}")
except PlanBlockedError as e:
    print(f"Plan blocked at step {e.blocked_step}: {e.reason}")
except SessionClosedError:
    print("Session was already closed")