Enterprise Prospecting & Outreach Orchestrator

from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions

options = ClaudeAgentOptions(
    system_prompt=(
        "You are the lead orchestrator. Use the Task tool to delegate to "
        "specialist subagents. Always maintain a 'case facts' block with "
        "non-negotiable transactional data. Before generating the research "
        "plan, ask 2-3 concise clarifying questions about outreach goals, "
        "including tone, competitor sensitivity, and technical depth."
    ),
    # The coordinator MUST have "Task" to spawn subagents.
    allowed_tools=["Task", "Read", "Grep", "mcp__crm__crm_query"],
    agents={
        "web-research": AgentDefinition(
            description=(
                "Researches a prospect's public footprint. Use for market "
                "signals, news, and technology-adoption research."
            ),
            prompt=(
                "Research the assigned prospect with the web_lookup tool. "
                "Return findings with sources; never speculate."
            ),
            tools=["mcp__research__web_lookup"],
            model="sonnet",
        ),
        "financial-analysis": AgentDefinition(
            description="Analyzes prospect financials from CRM data.",
            prompt=(
                "Analyze revenue, budget, and spend signals from CRM records. "
                "Return a structured summary with nullable fields for unknowns."
            ),
            tools=["mcp__crm__crm_query"],
            model="sonnet",
        ),
        "competitor-tracking": AgentDefinition(
            description="Tracks competitor positioning in a prospect's market.",
            prompt=(
                "Identify competitors and positioning gaps for the assigned "
                "market segment. Return findings with sources."
            ),
            tools=["mcp__research__web_lookup"],
            model="haiku",
        ),
    },
)

{
  "mcpServers": {
    "crm": {
      "command": "node",
      "args": ["./mcp-servers/crm-server.js"],
      "env": {
        "CRM_TOKEN": "${CRM_TOKEN}",
        "CRM_BASE_URL": "${CRM_BASE_URL}"
      }
    }
  }
}

{
  "resources": [
    {
      "uri": "crm://docs/prospecting",
      "name": "CRM prospecting documentation",
      "description": "Use before lead searches. Explains lead stages, required fields, and routing rules.",
      "mimeType": "text/markdown"
    },
    {
      "uri": "crm://schemas/leads",
      "name": "Leads schema",
      "description": "Use before CRM lead lookups. Lists column names, allowed enum values, and join keys so the agent chooses crm_query_leads instead of built-in Grep for structured account data.",
      "mimeType": "application/json"
    },
    {
      "uri": "crm://workflows/outreach-approval",
      "name": "Outreach approval workflow",
      "description": "Approval gates required before process_outreach can run.",
      "mimeType": "text/markdown"
    }
  ]
}

import anyio
from claude_agent_sdk import ClaudeSDKClient

async def run_research():
    # `options` is the ClaudeAgentOptions from Phase 1 (Task + AgentDefinitions)
    async with ClaudeSDKClient(options=options) as coordinator:
        await coordinator.query(
            "Research AI adoption in Fintech and Healthcare simultaneously. "
            "Delegate to the web-research, financial-analysis, and "
            "competitor-tracking subagents — issue one Task call per subagent, "
            "all in the same response, so they run in parallel. Each subagent "
            "starts with an empty context: include the prospect summary and "
            "the specific question in each Task prompt."
        )
        async for message in coordinator.receive_response():
            print(message)

anyio.run(run_research)

def subagent_tool_handler(tool_input):
    try:
        return {"data": run_research(tool_input)}
    except TimeoutError:
        return {
            "errorCategory": "TRANSIENT",
            "failure_type": "timeout",
            "isRetryable": True,
            "message": "Research API timed out. Retry with a narrower query."
        }
    except PermissionError:
        return {
            "errorCategory": "PERMISSION",
            "isRetryable": False,
            "message": "Token lacks access. Coordinator should pivot, not retry."
        }

aggregated_research = {
    "key_findings": [
        "Fintech CTOs show high interest in agentic workflow governance.",
        "Healthcare prospects require stricter compliance language and human review."
    ],
    "sections": {
        "prospect_startup_segment": startup_evidence,
        "prospect_fortune_500_segment": enterprise_evidence,
        "source_crm_notes": crm_notes,
        "source_web_research": cited_web_findings,
    },
}

# Plain single-shot generation requests: research is already done, so each
# request carries everything the model needs. No tools inside the batch.
message_batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"outreach-{prospect['crm_id']}",   # unique per prospect
            "params": {
                "model": "claude-sonnet-4-6",
                "max_tokens": 2048,
                "messages": [{
                    "role": "user",
                    "content": (
                        "Draft a personalized outreach email.\n"
                        f"Prospect brief: {prospect['brief']}\n"
                        f"Key findings: {prospect['key_findings']}\n"
                        "Follow the standard outreach template rules."
                    ),
                }],
            },
        }
        for prospect in qualified_prospects
    ]
)

# Later: correlate by custom_id — results arrive in any order.
results = {}
for result in client.messages.batches.results(message_batch.id):
    results[result.custom_id] = result.result

# Resubmit ONLY the failed/expired requests, never the whole batch.
retry_ids = [cid for cid, r in results.items() if r.type in ("errored", "expired")]

output_config = {
    "format": {
        "type": "json_schema",
        "schema": {
            "type": "object",
            "properties": {
                "company_name":   {"type": "string"},
                "contact_email":  {"type": "string"},
                "annual_revenue": {"type": ["integer", "null"]},
                "budget_range":   {"type": ["string", "null"]},
                "industry": {
                    "type": "string",
                    "enum": ["fintech", "healthcare", "manufacturing", "retail", "other"]
                },
                "industry_detail": {"type": ["string", "null"]},
                "reason_for_outreach": {
                    "type": "string",
                    "enum": ["ai_governance", "workflow_automation", "security_review", "other"]
                },
                "reason_detail": {"type": ["string", "null"]}
            },
            "required": [
                "company_name", "contact_email", "annual_revenue", "budget_range",
                "industry", "industry_detail", "reason_for_outreach", "reason_detail"
            ],
            "additionalProperties": False
        }
    }
}

CASE_FACTS = (
    "<case_facts>\n"
    "Max Budget: $50,000\n"
    "Contract term: 12 months\n"
    "Prospect opt-outs: no phone contact\n"
    "</case_facts>"
)

def build_turn(user_text: str) -> dict:
    # Re-inject case facts verbatim on EVERY turn so they can never
    # be paraphrased away as the conversation grows.
    return {
        "role": "user",
        "content": f"{CASE_FACTS}\n\n{user_text}",
    }

def export_state_manifest(path: str, state: dict) -> None:
    """Each agent writes its progress after every phase for crash recovery."""
    import json
    with open(path, "w") as f:
        json.dump({
            "completed_subtasks": state["done"],
            "pending_prospects": state["pending"],
            "verified_facts": state["facts"],
        }, f, indent=2)

{
  "hooks": {
    "PreToolUse": [
      {
        "matcher": {"tool_name": "process_outreach"},
        "hooks": [
          {
            "type": "command",
            "command": "python verify_risk_score.py"
          }
        ]
      }
    ]
  }
}

---
paths: ["outreach/**/*.md", "outreach/**/*.html"]
---
# Standard Outreach Template Rules
- Open with the prospect's confirmed pain point from CRM, never a generic hook.
- Include one quantified case study from the same industry vertical.
- Close with a single, specific call to action (no menu of options).
- Never reference compliance regimes the prospect has not opted into.

---
name: generate-audit
description: Audit generated outreach emails when asked to review campaign copy, regulated claims, unsupported personalization, or compliance-sensitive prospect messaging.
context: fork
allowed-tools: [Read, Grep, Glob]
---
# Audit Skill
1. Read the email under review.
2. Cross-check claims against the prospect's stated regulated regimes.
3. Open `references/compliance-claims.md` only when the email mentions HIPAA, SOC 2, ISO 27001, privacy, safety, or procurement claims.
4. Flag unsupported claims, weak personalization, policy conflicts, and missing escalation triggers.
5. Return a structured JSON report with findings, severity, evidence, and recommended next action.

# First run on a PR
claude -p "Review this capstone PR and return inline findings." \
  --output-format json \
  --json-schema .claude/schemas/pr-review-findings.schema.json

# Re-run after new commits: pass prior findings and dedupe
claude -p "Review this PR again. Prior findings: $(cat prior-findings.json). \
Report ONLY issues that are new or still unaddressed — do not repeat \
findings that were already posted or have been fixed." \
  --output-format json \
  --json-schema .claude/schemas/pr-review-findings.schema.json