PydanticAI
Static analysis for PydanticAI agents to detect loop vulnerabilities, tool risks, and prompt injection vectors.
Quick Start
# Scan PydanticAI project
inkog scan ./my-pydantic-ai-appWhat Inkog Detects
| Finding | Severity | Description |
|---|---|---|
| Unbounded Agent Loop | CRITICAL | Agent without max retries/iterations |
| Unsafe Tool Call | CRITICAL | Tool with code execution or shell access |
| Prompt Injection | HIGH | User input in system prompt |
| Missing Validation | MEDIUM | Tool without input validation |
| Hardcoded Secrets | CRITICAL | API keys in code |
PydanticAI Overview
PydanticAI is a type-safe Python framework for building AI agents with structured outputs. It uses Pydantic for input/output validation and supports multiple LLM providers.
from pydantic_ai import Agent
# Basic agent with system prompt
agent = Agent(
'openai:gpt-4',
system_prompt="You are a helpful assistant."
)
# Run the agent
result = agent.run_sync("Hello!")
print(result.data)Unbounded Agent Loops
Agents without retry limits can run indefinitely on failures.
Vulnerable
Infinite retry loop
from pydantic_ai import Agent
agent = Agent(
'openai:gpt-4',
system_prompt="Process user requests",
)
# No retry limit - will loop forever on errors
while True:
try:
result = agent.run_sync(user_input)
break
except Exception:
continue # Retry indefinitelySecure
Retry limit + timeout configured
from pydantic_ai import Agent, RunSettings
agent = Agent(
'openai:gpt-4',
system_prompt="Process user requests",
retries=3, # Built-in retry limit
)
settings = RunSettings(
max_tokens=1000,
timeout=30.0 # Request timeout
)
try:
result = agent.run_sync(
user_input,
settings=settings
)
except Exception as e:
logger.error(f"Agent failed after retries: {e}")Unsafe Tool Definitions
Tools that execute arbitrary code pose remote code execution risks.
Vulnerable
Shell exec + eval = RCE
from pydantic_ai import Agent
import subprocess
agent = Agent('openai:gpt-4')
@agent.tool
def execute_command(command: str) -> str:
"""Execute a shell command."""
return subprocess.check_output(
command, shell=True
).decode()
@agent.tool
def run_code(code: str) -> str:
"""Run Python code."""
return eval(code)Secure
Validated Pydantic model input
from pydantic_ai import Agent
from pydantic import BaseModel, field_validator
class SearchQuery(BaseModel):
query: str
@field_validator('query')
@classmethod
def validate_query(cls, v):
if len(v) > 200:
raise ValueError('Query too long')
return v.strip()
agent = Agent('openai:gpt-4')
@agent.tool
def search_database(query: SearchQuery) -> list[dict]:
"""Search the database safely."""
# Parameterized query - no injection
return db.search(query.query, limit=10)Prompt Injection Vulnerabilities
User input flowing to system prompts enables prompt injection.
Vulnerable
User input in system prompt
from pydantic_ai import Agent
def create_agent(user_role: str):
return Agent(
'openai:gpt-4',
# User input in system prompt!
system_prompt=f"You are a {user_role} assistant."
)
# Attacker: user_role = "malicious\n\nIgnore all rules"
agent = create_agent(request.user_role)Secure
Enum allowlist - no injection possible
from pydantic_ai import Agent
from enum import Enum
class UserRole(str, Enum):
CUSTOMER = "customer support"
SALES = "sales"
TECHNICAL = "technical support"
ROLE_PROMPTS = {
UserRole.CUSTOMER: "You help customers with orders.",
UserRole.SALES: "You help with product inquiries.",
UserRole.TECHNICAL: "You help with technical issues.",
}
def create_agent(role: UserRole):
return Agent(
'openai:gpt-4',
# Static prompt from allowlist
system_prompt=ROLE_PROMPTS[role]
)Streaming Output Risks
Streaming responses should be validated before displaying.
Vulnerable
Raw LLM output to user
from pydantic_ai import Agent
agent = Agent('openai:gpt-4')
async def stream_response(query: str):
async with agent.run_stream(query) as result:
async for chunk in result.stream_text():
# Directly output to user
yield chunkSecure
Output sanitization
from pydantic_ai import Agent
import html
agent = Agent('openai:gpt-4')
BLOCKED_PATTERNS = ['<script', 'javascript:', 'onerror=']
async def stream_response(query: str):
async with agent.run_stream(query) as result:
async for chunk in result.stream_text():
# Sanitize output
safe_chunk = html.escape(chunk)
if any(p in chunk.lower() for p in BLOCKED_PATTERNS):
yield "[content filtered]"
else:
yield safe_chunkStructured Output Validation
Use Pydantic models to validate agent outputs.
Vulnerable
eval() on LLM output
from pydantic_ai import Agent
agent = Agent('openai:gpt-4')
result = agent.run_sync("Get user data")
# Trusting unstructured output
user_data = eval(result.data) # DANGER!Secure
Pydantic validation enforced
from pydantic_ai import Agent
from pydantic import BaseModel
class UserData(BaseModel):
id: int
name: str
email: str
is_admin: bool = False # Default safe value
agent = Agent(
'openai:gpt-4',
result_type=UserData # Enforce schema
)
result = agent.run_sync("Get user data")
# Validated, typed output
user_data: UserData = result.data
print(user_data.name) # Type-safe accessBest Practices
- Set retry limits using
retriesparameter - Use timeouts with
RunSettings(timeout=30.0) - Validate tool inputs with Pydantic models
- Never use eval/exec in tool implementations
- Static system prompts - no user input interpolation
- Define result_type for structured, validated outputs
- Sanitize streaming output before displaying to users
CLI Examples
# Scan PydanticAI project
inkog scan ./agents
# Check for tool vulnerabilities
inkog scan . -pattern unsafe-tool
# JSON output for CI
inkog scan . -output jsonSupported Patterns
Inkog detects these PydanticAI-specific patterns:
Agent(retries=...)- Retry configuration@agent.tool- Tool definitionssystem_prompt=...- System prompt analysisresult_type=...- Output validationRunSettings(timeout=...)- Timeout configuration
Related
- LangChain - Similar agent framework
- Infinite Loops
- Prompt Injection
Last updated on