Skip to content

Commit 3173785

Browse files
VladUZHclaude
andcommitted
docs: add Pydantic AI integration example
Complete the examples directory with pydantic_ai_example.py showing governance_dependency usage within Pydantic AI tool functions. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 1c55780 commit 3173785

1 file changed

Lines changed: 115 additions & 0 deletions

File tree

examples/pydantic_ai_example.py

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
"""Pydantic AI integration example.
2+
3+
Demonstrates using SidClaw governance within Pydantic AI tool functions
4+
via the governance_dependency helper.
5+
6+
Usage:
7+
pip install sidclaw[pydantic-ai] pydantic-ai
8+
export SIDCLAW_API_KEY=ai_...
9+
export SIDCLAW_AGENT_ID=your-agent-id
10+
export OPENAI_API_KEY=sk-...
11+
python pydantic_ai_example.py
12+
"""
13+
14+
import os
15+
16+
from pydantic_ai import Agent, RunContext
17+
18+
from sidclaw import ActionDeniedError, AsyncSidClaw
19+
from sidclaw.middleware.pydantic_ai import governance_dependency
20+
21+
22+
# --- SidClaw client ---
23+
24+
client = AsyncSidClaw(
25+
api_key=os.environ["SIDCLAW_API_KEY"],
26+
agent_id=os.environ["SIDCLAW_AGENT_ID"],
27+
)
28+
29+
30+
# --- Dependencies ---
31+
32+
class Deps:
33+
sidclaw_client: AsyncSidClaw = client
34+
35+
36+
# --- Agent ---
37+
38+
agent = Agent("openai:gpt-4", deps_type=Deps)
39+
40+
41+
@agent.tool
42+
async def lookup_customer(ctx: RunContext[Deps], email: str) -> str:
43+
"""Look up a customer by email address."""
44+
gov = governance_dependency(ctx.deps.sidclaw_client)
45+
decision = await gov.check(
46+
"lookup_customer",
47+
target_integration="crm",
48+
data_classification="confidential",
49+
)
50+
51+
try:
52+
# Your CRM lookup logic here
53+
result = f"Customer found: {email} (Enterprise plan)"
54+
await gov.record_success(decision.trace_id)
55+
return result
56+
except Exception as e:
57+
await gov.record_error(decision.trace_id, e)
58+
raise
59+
60+
61+
@agent.tool
62+
async def send_email(ctx: RunContext[Deps], to: str, subject: str, body: str) -> str:
63+
"""Send an email to a customer."""
64+
gov = governance_dependency(ctx.deps.sidclaw_client)
65+
66+
try:
67+
decision = await gov.check(
68+
"send_email",
69+
target_integration="email_service",
70+
data_classification="confidential",
71+
)
72+
except ActionDeniedError as e:
73+
return f"Email blocked by policy: {e.reason}"
74+
75+
try:
76+
# Your email sending logic here
77+
result = f"Email sent to {to}: {subject}"
78+
await gov.record_success(decision.trace_id)
79+
return result
80+
except Exception as e:
81+
await gov.record_error(decision.trace_id, e)
82+
raise
83+
84+
85+
@agent.tool
86+
async def export_data(ctx: RunContext[Deps], query: str) -> str:
87+
"""Export customer data (blocked by policy)."""
88+
gov = governance_dependency(ctx.deps.sidclaw_client)
89+
90+
try:
91+
decision = await gov.check(
92+
"export_data",
93+
target_integration="crm",
94+
data_classification="restricted",
95+
)
96+
except ActionDeniedError as e:
97+
return f"Data export blocked: {e.reason}"
98+
99+
# This won't be reached if policy denies
100+
return "Export complete"
101+
102+
103+
# --- Run ---
104+
105+
async def main():
106+
result = await agent.run(
107+
"Look up the customer john@example.com and send them a follow-up email",
108+
deps=Deps(),
109+
)
110+
print(result.data)
111+
112+
113+
if __name__ == "__main__":
114+
import asyncio
115+
asyncio.run(main())

0 commit comments

Comments
 (0)