Loading lesson…
Loading lesson…
The previous scenarios all used one agent for one job. In real enterprise environments, what you need is a full pipeline - multiple agents chained together, each stage with clear input/output, plus auditing, rollback, and notifications.
Now we will build a real scenario: after each PR submission, automatically trigger code review -> security scan -> auto-fix -> test verification -> report generation as a complete pipeline.
PR submitted
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Code Review │───▶│ Security Scan│───▶│ Auto Fix │
│ Agent │ │ Agent │ │ Agent │
│ (read-only) │ │ (read-only) │ │ (writable) │
└─────────────┘ └─────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌─────────────┐
│ Test Verify │───▶│ Report Build │
│ Agent │ │ Agent │
│ (Bash) │ │ (Write) │
└─────────────┘ └─────────────┘
│
▼
Slack notification
Core idea: each agent does one thing, permissions are minimized, and results are passed in sequence.
import asyncio
import json
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher
# Audit log: record every operation by every agent
audit_log = []
async def audit_hook(input_data, tool_use_id, context):
audit_log.append({
"time": datetime.now().isoformat(),
"tool": input_data.get("tool_name"),
"input": input_data.get("tool_input", {}),
})
return {}
# Shared hook config: all agents share audit capability
audit_hooks = {
"PostToolUse": [HookMatcher(matcher=".*", hooks=[audit_hook])]
}
async def run_code_review(pr_diff: str) -> str:
"""Read-only agent, reviews code quality and outputs a structured report"""
result_text = ""
async for message in query(
prompt=f"""Review the following PR diff from these dimensions:
1. Code conventions: naming, formatting, comments
2. Logic issues: edge cases, null pointer risks, race conditions
3. Performance risks: N+1 queries, memory leaks, unnecessary loops
4. Maintainability: oversized functions, unclear responsibilities, magic numbers
PR Diff:
{pr_diff}
Output JSON format: {{"issues": [{{"severity": "high/medium/low", "file": "...", "line": ..., "description": "..."}}], "summary": "..."}}""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep"],
permission_mode="bypassPermissions",
hooks=audit_hooks,
max_turns=10,
),
):
if hasattr(message, "result"):
result_text = message.result
return result_text
async def run_security_scan() -> str:
"""Read-only agent focused on vulnerability scanning"""
result_text = ""
async for message in query(
prompt="""Scan the project code for security vulnerabilities:
1. SQL injection, XSS, CSRF
2. Hardcoded keys or credentials
3. Insecure dependency versions
4. Missing permission checks
Output JSON: {{"vulnerabilities": [{{"severity": "critical/high/medium", "type": "...", "file": "...", "description": "...", "fix_suggestion": "..."}}]}}""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep", "Bash"],
permission_mode="bypassPermissions",
hooks=audit_hooks,
max_turns=15,
),
):
if hasattr(message, "result"):
result_text = message.result
return result_text
async def run_auto_fix(review_result: str, security_result: str) -> str:
"""Writable agent that auto-fixes code based on review and scan results"""
result_text = ""
async for message in query(
prompt=f"""Fix code according to the following review results:
Code review report:
{review_result}
Security scan report:
{security_result}
Fix rules:
1. Only fix issues with severity high or critical
2. Run related tests after each change to ensure no existing functionality is broken
3. Do not refactor unrelated code, apply minimal fixes only
4. Output the list of modified files after completion""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit", "Bash", "Glob", "Grep"],
permission_mode="acceptEdits",
hooks=audit_hooks,
max_turns=30,
),
):
if hasattr(message, "result"):
result_text = message.result
return result_text
async def run_test_and_report(fix_result: str) -> str:
"""Run tests and generate final report"""
result_text = ""
async for message in query(
prompt=f"""Execute these actions:
1. Run the full test suite (npm test or pytest)
2. Compute test pass rate
3. Generate a Markdown quality report into pr-report.md, including:
- Count of issues found in code review and severity distribution
- Number of security vulnerabilities
- Auto-fix changes: {fix_result}
- Test pass rate
- Final conclusion: whether merge is recommended""",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Bash", "Write", "Glob"],
hooks=audit_hooks,
max_turns=15,
),
):
if hasattr(message, "result"):
result_text = message.result
return result_text
import subprocess
async def run_pipeline():
"""Full PR quality-guard pipeline"""
print("🔍 Stage 1/4: code review...")
pr_diff = subprocess.run(
["git", "diff", "main...HEAD"], capture_output=True, text=True
).stdout
review_result = await run_code_review(pr_diff)
print("🛡️ Stage 2/4: security scan...")
security_result = await run_security_scan()
print("🔧 Stage 3/4: auto-fix...")
fix_result = await run_auto_fix(review_result, security_result)
print("✅ Stage 4/4: test verification + report generation...")
report = await run_test_and_report(fix_result)
# Save audit log
with open("audit-log.json", "w") as f:
json.dump(audit_log, f, indent=2, ensure_ascii=False)
print(f"Pipeline finished, audit log saved ({len(audit_log)} operation records)")
return report
asyncio.run(run_pipeline())
This pipeline reflects several key enterprise design principles:
Least privilege: code-review and security-scan agents are read-only and cannot accidentally modify code. Only the auto-fix agent has write permission, and even that is constrained by acceptEdits.
Auditable: every step of every agent is logged through Hooks. If anything goes wrong, you can trace which agent did what and when.
Result chaining: each agent's output becomes the next agent's input. Review results feed auto-fix; auto-fix results feed test verification. Every stage has a clear input/output contract.
Cost control: every agent has a max_turns limit to prevent runaway loops. In production, you can also add max_budget_usd for budget control.
Extensibility: want another stage, such as a "documentation-check agent" or "performance benchmark agent"? Add a new function and insert it into the pipeline.
This model can be embedded directly into GitHub Actions or GitLab CI, automatically triggered on each PR, truly achieving "AI-driven code quality guardrails."
Agent SDK provides clear exception types so you can build robust fault tolerance in production:
from claude_agent_sdk import query, CLINotFoundError, ProcessError
try:
async for msg in query(prompt="Analyze code"):
print(msg)
except CLINotFoundError:
print("Claude Code CLI is not installed. Please install it first.")
except ProcessError as e:
print(f"Process exited unexpectedly with exit code: {e.exit_code}")
The core value of Claude Agent SDK is upgrading "model reasoning" into "controlled execution." It does not just generate text. It can truly complete tasks inside an auditable, constrained tool system.
Remember a line from Anthropic's official blog: the Agent SDK design philosophy is "give the agent a computer and let it work like a human."
A good agent application = clear tool design + explicit task boundaries + appropriate human supervision. Tools give the agent capability, boundaries give it constraints, and supervision gives you confidence. None of the three can be missing.