Deploy Your Agent — Local, GCP & AWS

mkdir ucc-agent-deploy && cd ucc-agent-deploy
python -m venv venv
# Linux/macOS:
source venv/bin/activate
# Windows:
# venv\Scripts\activate

pip install anthropic>=0.30.0 fastapi>=0.115.0 uvicorn>=0.34.0 mangum>=0.19.0 pydantic>=2.0.0
pip freeze > requirements.txt

# Set your API key (never hardcode this in files!)
export ANTHROPIC_API_KEY=your-key-here
# Windows: set ANTHROPIC_API_KEY=your-key-here

# Copy this file to .env and fill in your values
# NEVER commit .env to git!
ANTHROPIC_API_KEY=your-anthropic-api-key-here
PORT=8000

venv/
node_modules/
__pycache__/
*.pyc
.env
.git/
*.md

# mock_data.py — Simulated UCC filing records
# In production, these would come from a database or API

FILINGS = [
    {
        "filing_id": "UCC-2024-NY-001847",
        "debtor_name": "Acme Corporation",
        "debtor_state": "NY",
        "secured_party": "First National Bank",
        "collateral_description": "All inventory, equipment, and accounts receivable",
        "filing_date": "2024-03-15",
        "status": "active",
        "collateral_value": 2500000,
    },
    {
        "filing_id": "UCC-2024-NY-002103",
        "debtor_name": "Acme Corporation",
        "debtor_state": "NY",
        "secured_party": "TechLease Partners LLC",
        "collateral_description": "Specific equipment: CNC machines, serial #TL-8842, #TL-8843",
        "filing_date": "2024-06-01",
        "status": "active",
        "collateral_value": 450000,
    },
    {
        "filing_id": "UCC-2023-CA-009821",
        "debtor_name": "Pacific Freight Inc",
        "debtor_state": "CA",
        "secured_party": "West Coast Capital",
        "collateral_description": "All assets including rolling stock and warehouse inventory",
        "filing_date": "2023-11-20",
        "status": "active",
        "collateral_value": 8700000,
    },
    {
        "filing_id": "UCC-2024-TX-004210",
        "debtor_name": "Lone Star Drilling Co",
        "debtor_state": "TX",
        "secured_party": "Energy Finance Corp",
        "collateral_description": "Drilling equipment, mineral rights assignments, accounts",
        "filing_date": "2024-01-10",
        "status": "active",
        "collateral_value": 15000000,
    },
    {
        "filing_id": "UCC-2022-NY-000412",
        "debtor_name": "Acme Corporation",
        "debtor_state": "NY",
        "secured_party": "Metro Business Lending",
        "collateral_description": "Accounts receivable",
        "filing_date": "2022-08-05",
        "status": "terminated",
        "collateral_value": 750000,
    },
]


def search_filings(debtor_name: str) -> list[dict]:
    """Search filings by debtor name (case-insensitive partial match)."""
    name_lower = debtor_name.lower()
    return [f for f in FILINGS if name_lower in f["debtor_name"].lower()]


def calculate_risk_score(debtor_name: str) -> dict:
    """Calculate a risk score based on filing history."""
    matches = search_filings(debtor_name)
    active = [f for f in matches if f["status"] == "active"]
    total_exposure = sum(f["collateral_value"] for f in active)

    if not matches:
        return {
            "debtor_name": debtor_name,
            "risk_level": "unknown",
            "risk_score": 0,
            "reason": "No filings found for this entity",
            "total_liens": 0,
            "total_exposure": 0,
        }

    if total_exposure > 10000000:
        risk_level, score = "high", 85
    elif total_exposure > 2000000:
        risk_level, score = "medium", 55
    elif len(active) > 1:
        risk_level, score = "medium", 45
    else:
        risk_level, score = "low", 20

    return {
        "debtor_name": debtor_name,
        "risk_level": risk_level,
        "risk_score": score,
        "reason": f"{len(active)} active lien(s) totaling ${total_exposure:,.0f}",
        "total_liens": len(active),
        "total_exposure": total_exposure,
    }

# agent.py — UCC Filing Research Agent
# Uses the Anthropic Messages API with tool use (M12 pattern)

import json
import os
import anthropic
from mock_data import search_filings, calculate_risk_score

# --- Tool definitions tell Claude what functions are available ---
# Each tool has a name, description (crucial for selection accuracy),
# and an input_schema that Claude uses to generate valid arguments.

TOOLS = [
    {
        "name": "search_filings",
        "description": (
            "Search UCC filings by debtor name. Returns a list of filing "
            "records including filing ID, secured party, collateral "
            "description, filing date, status, and collateral value. "
            "Use this to find all liens against a specific company."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "debtor_name": {
                    "type": "string",
                    "description": "The company or individual name to search for (partial match supported)",
                }
            },
            "required": ["debtor_name"],
        },
    },
    {
        "name": "get_risk_score",
        "description": (
            "Calculate a lien risk score for a debtor based on their UCC "
            "filing history. Returns risk level (low/medium/high), numeric "
            "score (0-100), total active liens, and total collateral exposure. "
            "Use this after searching filings to assess overall risk."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "debtor_name": {
                    "type": "string",
                    "description": "The debtor name to assess risk for",
                }
            },
            "required": ["debtor_name"],
        },
    },
]

# --- Map tool names to Python functions ---
TOOL_HANDLERS = {
    "search_filings": lambda args: search_filings(args["debtor_name"]),
    "get_risk_score": lambda args: calculate_risk_score(args["debtor_name"]),
}


def run_agent(question: str, max_turns: int = 10) -> str:
    """
    Run the UCC research agent with a tool use loop.

    The loop checks stop_reason after each Claude response:
    - "tool_use" means Claude wants to call a tool — execute it and continue
    - "end_turn" means Claude is done — return the final text

    max_turns is a safety net to prevent infinite loops, not a control mechanism.
    """
    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from environment

    messages = [{"role": "user", "content": question}]

    system_prompt = (
        "You are a UCC filing research assistant. You help users investigate "
        "Uniform Commercial Code filings, assess lien risk, and understand "
        "collateral exposure for business entities. Always search for filings "
        "first, then calculate risk scores to give comprehensive answers. "
        "Present findings clearly with specific numbers and filing IDs."
    )

    for turn in range(max_turns):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                system=system_prompt,
                tools=TOOLS,
                messages=messages,
            )
        except anthropic.APIError as e:
            return f"API error: {e.message}"
        except anthropic.AuthenticationError:
            return "Authentication failed. Check your ANTHROPIC_API_KEY."

        # Check if Claude wants to use tools or is done
        if response.stop_reason == "end_turn":
            # Extract the text from the response
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "Agent completed but produced no text output."

        if response.stop_reason == "tool_use":
            # Append Claude's response (which includes tool_use blocks)
            messages.append({"role": "assistant", "content": response.content})

            # Execute each tool call and collect results
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    handler = TOOL_HANDLERS.get(block.name)
                    if handler:
                        try:
                            result = handler(block.input)
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": json.dumps(result, default=str),
                            })
                        except Exception as exc:
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": json.dumps({
                                    "error": str(exc),
                                    "is_retryable": False,
                                }),
                                "is_error": True,
                            })
                    else:
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps({
                                "error": f"Unknown tool: {block.name}",
                                "is_retryable": False,
                            }),
                            "is_error": True,
                        })

            messages.append({"role": "user", "content": tool_results})

    return "Agent reached maximum turns without completing. Try a simpler question."


if __name__ == "__main__":
    # Quick test — run directly to verify the agent works
    answer = run_agent("What is the lien exposure for Acme Corporation?")
    print(answer)

# server.py — FastAPI wrapper for the UCC research agent
# Exposes the agent as a REST API with health check

import os
import time
import logging
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

from agent import run_agent

# --- Logging setup ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ucc-agent-api")

# --- Request/Response models ---
# Pydantic validates every incoming request automatically.
# If a client sends {"query": 123} instead of a string, they get
# a clear 422 error before the agent ever runs.

class QueryRequest(BaseModel):
    question: str = Field(
        ...,
        min_length=1,
        max_length=2000,
        description="The question to ask the UCC research agent",
        json_schema_extra={"examples": ["Find filings for Acme Corporation"]},
    )

class QueryResponse(BaseModel):
    answer: str
    elapsed_seconds: float
    status: str = "success"

class HealthResponse(BaseModel):
    status: str
    service: str
    version: str

class ErrorResponse(BaseModel):
    detail: str
    status: str = "error"

# --- Application setup ---
app = FastAPI(
    title="UCC Filing Research Agent API",
    description="AI-powered UCC filing research and lien risk assessment",
    version="1.0.0",
)

# Allow browser-based clients to call this API
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict in production!
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Track request count for basic metrics
request_count = 0
error_count = 0
start_time = time.time()


@app.get("/health", response_model=HealthResponse, tags=["System"])
async def health_check():
    """Health check endpoint for load balancers and container orchestration."""
    # Check that the API key is configured (do not reveal the key!)
    api_key = os.environ.get("ANTHROPIC_API_KEY", "")
    if not api_key:
        raise HTTPException(
            status_code=503,
            detail="ANTHROPIC_API_KEY not configured",
        )
    return HealthResponse(
        status="healthy",
        service="ucc-agent",
        version="1.0.0",
    )


@app.post(
    "/query",
    response_model=QueryResponse,
    responses={500: {"model": ErrorResponse}},
    tags=["Agent"],
)
async def query_agent(request: QueryRequest):
    """Send a question to the UCC research agent and get an answer."""
    global request_count, error_count
    request_count += 1

    logger.info(f"Query received: {request.question[:80]}...")
    t0 = time.time()

    try:
        answer = run_agent(request.question)
        elapsed = round(time.time() - t0, 2)
        logger.info(f"Query completed in {elapsed}s")
        return QueryResponse(
            answer=answer,
            elapsed_seconds=elapsed,
        )
    except Exception as exc:
        error_count += 1
        logger.error(f"Agent error: {exc}")
        raise HTTPException(
            status_code=500,
            detail=f"Agent failed: {str(exc)}",
        )


@app.get("/metrics", tags=["System"])
async def metrics():
    """Basic metrics endpoint for monitoring."""
    uptime = round(time.time() - start_time, 1)
    return {
        "total_requests": request_count,
        "total_errors": error_count,
        "uptime_seconds": uptime,
        "error_rate": round(error_count / max(request_count, 1), 4),
    }


if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 8000))
    uvicorn.run(app, host="0.0.0.0", port=port)

# Test 1: Health check
curl -s http://localhost:8000/health | python -m json.tool

# Test 2: Query the agent
curl -s -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "Find filings for Acme Corporation"}' | python -m json.tool

# Test 3: Metrics
curl -s http://localhost:8000/metrics | python -m json.tool

# Test 4: Invalid request (should get 422)
curl -s -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": ""}' | python -m json.tool

# Dockerfile — Multi-stage build for the UCC agent

# ----- Stage 1: Build dependencies in a full Python image -----
# WHAT: Use the full Python image to install compiled dependencies
# WHY: Some pip packages (like uvloop) need C compilers to build.
#       The slim image does not have compilers, so we build here first.
FROM python:3.12 AS builder

WORKDIR /build

# WHAT: Copy only requirements.txt first, then install
# WHY: Docker caches each layer. If requirements.txt hasn't changed,
#       Docker reuses the cached pip install (saves 30-60 seconds).
# GOTCHA: If you COPY . . first, ANY code change invalidates the cache.
COPY requirements.txt .
RUN pip install --no-cache-dir --target=/build/deps -r requirements.txt

# ----- Stage 2: Slim runtime image -----
# WHAT: Start fresh with a minimal Python image
# WHY: The builder image is ~900MB (compilers, headers). The slim image
#       is ~150MB. We only copy the installed packages, not the compilers.
FROM python:3.12-slim

# WHAT: Create a non-root user
# WHY: Running as root inside a container is a security risk. If an
#       attacker exploits a bug, they get root access to the container.
#       A non-root user limits the damage.
RUN useradd --create-home --shell /bin/bash agent

WORKDIR /app

# WHAT: Copy installed dependencies from the builder stage
# WHY: We get all the pip packages without the build tools
COPY --from=builder /build/deps /usr/local/lib/python3.12/site-packages/

# WHAT: Copy application source code
# WHY: This layer changes most often (your code changes), so it goes LAST
#       to maximize cache hits on the layers above.
COPY mock_data.py agent.py server.py ./

# WHAT: Switch to the non-root user
USER agent

# WHAT: Expose port 8000 and set the default command
# WHY: EXPOSE documents which port the container listens on.
#       CMD sets what runs when the container starts.
# GOTCHA: EXPOSE does not publish the port — you still need -p 8000:8000
EXPOSE 8000
ENV PORT=8000

# WHAT: Run uvicorn with production settings
# WHY: --host 0.0.0.0 makes it accessible from outside the container.
#       --workers 1 is fine for agent workloads (they are I/O bound,
#       waiting on Claude API, not CPU bound).
CMD ["python", "-m", "uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

# Build the image (run from the project directory)
docker build -t ucc-agent .

# Run the container, passing the API key from your environment
docker run -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY ucc-agent

# Or on Windows PowerShell:
# docker run -p 8000:8000 -e ANTHROPIC_API_KEY=$env:ANTHROPIC_API_KEY ucc-agent

curl -s http://localhost:8000/health | python -m json.tool

curl -s -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the lien exposure for Acme Corporation?"}' \
  | python -m json.tool

# docker-compose.yml — Development orchestration
version: "3.8"

services:
  agent:
    build: .
    ports:
      - "8000:8000"
    env_file:
      - .env                    # Reads ANTHROPIC_API_KEY from .env file
    environment:
      - PORT=8000
    # Uncomment the next 2 lines for hot-reload during development:
    # volumes:
    #   - ./:/app               # Mount source code into container
    restart: unless-stopped     # Auto-restart on crash
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 10s

ANTHROPIC_API_KEY=your-actual-api-key-here
PORT=8000

# 1. Authenticate (opens browser for Google login)
gcloud auth login

# 2. Set your project (replace with your GCP project ID)
gcloud config set project YOUR_PROJECT_ID

# 3. Enable the required APIs
gcloud services enable run.googleapis.com artifactregistry.googleapis.com

# 4. Create an Artifact Registry repository for Docker images
gcloud artifacts repositories create agents \
  --repository-format=docker \
  --location=us-central1 \
  --description="Agent container images"

# 5. Configure Docker to authenticate with Artifact Registry
gcloud auth configure-docker us-central1-docker.pkg.dev

# 6. Verify your setup
gcloud config list

# 1. Tag the image for Artifact Registry
docker tag ucc-agent \
  us-central1-docker.pkg.dev/YOUR_PROJECT_ID/agents/ucc-agent:v1

# 2. Push the image
docker push us-central1-docker.pkg.dev/YOUR_PROJECT_ID/agents/ucc-agent:v1

# 3. Deploy to Cloud Run
# Every flag explained:
#   --image       : the container image to run
#   --region      : where to run (us-central1 is cheap, low latency for US)
#   --set-env-vars: pass the API key (for production, use Secret Manager instead)
#   --memory 512Mi: 512 MB RAM — enough for an agent + Python runtime
#   --cpu 1       : 1 vCPU — agents are I/O bound, not CPU bound
#   --timeout 60  : 60-second request timeout — agent loops can take 10-30s
#   --max-instances 3: cost safety net — prevents scaling to 1000 instances
#                       from a traffic spike (each instance = money)
#   --allow-unauthenticated: makes the URL public (remove for production)

gcloud run deploy ucc-agent \
  --image us-central1-docker.pkg.dev/YOUR_PROJECT_ID/agents/ucc-agent:v1 \
  --region us-central1 \
  --set-env-vars ANTHROPIC_API_KEY=your-key-here \
  --memory 512Mi \
  --cpu 1 \
  --timeout 60 \
  --max-instances 3 \
  --allow-unauthenticated

# Replace with YOUR actual service URL from the deploy output
SERVICE_URL="https://ucc-agent-abc123-uc.a.run.app"

# Health check
curl -s $SERVICE_URL/health | python -m json.tool

# Query the agent
curl -s -X POST $SERVICE_URL/query \
  -H "Content-Type: application/json" \
  -d '{"question": "Find all filings for Pacific Freight Inc"}' \
  | python -m json.tool

# 1. Configure AWS CLI (you will be prompted for access key, secret, region)
aws configure
# Access Key ID: your-access-key
# Secret Access Key: your-secret-key
# Default region: us-east-1
# Default output format: json

# 2. Verify your identity
aws sts get-caller-identity

# 3. Verify SAM CLI is installed
sam --version

# lambda_handler.py — AWS Lambda adapter for FastAPI
# Mangum converts API Gateway events into ASGI requests that FastAPI understands.

from mangum import Mangum
from server import app

# This is the entry point Lambda calls.
# Mangum wraps the FastAPI app: Lambda event -> HTTP request -> FastAPI -> HTTP response -> Lambda response
handler = Mangum(app, lifespan="off")

# That's it! Mangum does all the translation. Your FastAPI routes,
# Pydantic validation, error handling — everything works exactly
# the same as it does locally. The only difference is HOW the
# request arrives (Lambda event vs. HTTP request).

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: UCC Filing Research Agent - Lambda deployment

Globals:
  Function:
    Timeout: 120           # 120 seconds - agent loops can take 10-30s
    MemorySize: 512        # 512 MB - enough for Python + anthropic SDK
    Runtime: python3.12

Resources:
  UCCAgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda_handler.handler
      CodeUri: .
      Description: UCC Filing Research Agent
      Architectures:
        - x86_64
      Environment:
        Variables:
          ANTHROPIC_API_KEY: !Ref AnthropicApiKey
      Events:
        HealthCheck:
          Type: Api
          Properties:
            Path: /health
            Method: get
        Query:
          Type: Api
          Properties:
            Path: /query
            Method: post
        Metrics:
          Type: Api
          Properties:
            Path: /metrics
            Method: get

Parameters:
  AnthropicApiKey:
    Type: String
    NoEcho: true           # Masks the value in CloudFormation logs
    Description: Anthropic API key for Claude access

Outputs:
  ApiUrl:
    Description: API Gateway URL for the UCC Agent
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/"

# Build the Lambda package (installs deps, creates deployment artifact)
sam build

# Deploy (first time: use --guided for interactive prompts)
sam deploy --guided
# Stack name: ucc-agent-stack
# Region: us-east-1
# Parameter AnthropicApiKey: (paste your API key — it will be masked)
# Confirm changes before deploy: y
# Allow SAM CLI IAM role creation: y
# Save arguments to samconfig.toml: y

# After first deploy, subsequent deploys are simpler:
# sam build && sam deploy

# Replace with YOUR API Gateway URL from the deploy output
LAMBDA_URL="https://abc123def4.execute-api.us-east-1.amazonaws.com/Prod"

curl -s $LAMBDA_URL/health | python -m json.tool

curl -s -X POST $LAMBDA_URL/query \
  -H "Content-Type: application/json" \
  -d '{"question": "What is the risk level for Lone Star Drilling Co?"}' \
  | python -m json.tool

# Docker: view container logs
docker logs -f $(docker ps -q --filter ancestor=ucc-agent)

# GCP Cloud Run: stream logs
gcloud run services logs read ucc-agent --region us-central1

# AWS Lambda: tail CloudWatch logs
sam logs --stack-name ucc-agent-stack --tail

# All platforms: check the /metrics endpoint
curl -s http://localhost:8000/metrics | python -m json.tool

QUERY='{"question": "What is the lien exposure for Acme Corporation?"}'

echo "=== Local Docker ==="
curl -s -X POST http://localhost:8000/query \
  -H "Content-Type: application/json" -d "$QUERY" | python -m json.tool

echo ""
echo "=== GCP Cloud Run ==="
curl -s -X POST https://ucc-agent-abc123-uc.a.run.app/query \
  -H "Content-Type: application/json" -d "$QUERY" | python -m json.tool

echo ""
echo "=== AWS Lambda ==="
curl -s -X POST https://abc123def4.execute-api.us-east-1.amazonaws.com/Prod/query \
  -H "Content-Type: application/json" -d "$QUERY" | python -m json.tool

import boto3

# Control-plane client: creates and configures agents
ctrl = boto3.client("bedrock-agent", region_name="us-west-2")

# Step 1: create the agent shell (system prompt + foundation model)
agent = ctrl.create_agent(
    agentName="ucc-research-agent",
    foundationModel="anthropic.claude-opus-4-7-v1:0",
    instruction=(
        "You are a UCC filing research assistant. Use the lookup_filings "
        "tool to search public records. Always cite filing numbers."
    ),
    agentResourceRoleArn="arn:aws:iam::123456789012:role/AgentExecutionRole",
    idleSessionTTLInSeconds=600,  # session memory expires after 10 min idle
)
agent_id = agent["agent"]["agentId"]

# Step 2: attach an action group (your tools, defined as a Lambda function)
ctrl.create_agent_action_group(
    agentId=agent_id,
    agentVersion="DRAFT",
    actionGroupName="filing-tools",
    actionGroupExecutor={
        "lambda": "arn:aws:lambda:us-west-2:123456789012:function:lookup-filings"
    },
    apiSchema={
        "s3": {
            "s3BucketName": "my-agent-schemas",
            "s3ObjectKey": "filing-tools-openapi.yaml",  # OpenAPI 3 spec
        }
    },
)

# Step 3: prepare and create an alias (the stable endpoint your clients call)
ctrl.prepare_agent(agentId=agent_id)
alias = ctrl.create_agent_alias(agentId=agent_id, agentAliasName="prod")
print(f"Agent ready: agentId={agent_id}, aliasId={alias['agentAlias']['agentAliasId']}")

import boto3, uuid

# Data-plane client: invokes deployed agents
rt = boto3.client("bedrock-agent-runtime", region_name="us-west-2")

def ask(question: str, session_id: str) -> str:
    response = rt.invoke_agent(
        agentId="ABCDE12345",
        agentAliasId="PRODALIAS1",   # the alias you created above
        sessionId=session_id,         # same id across calls = same conversation
        inputText=question,
        enableTrace=True,             # returns reasoning steps for debugging
    )
    # invoke_agent streams; concatenate chunks for the final answer
    answer = []
    for event in response["completion"]:
        if "chunk" in event:
            answer.append(event["chunk"]["bytes"].decode("utf-8"))
        elif "trace" in event:
            # Useful for production logs: which tool was called, why
            print("trace:", event["trace"])
    return "".join(answer)

# Same session_id = Bedrock automatically maintains conversation memory.
# No Redis. No conversations table. The platform owns it.
session = str(uuid.uuid4())
print(ask("What were Q4 sales for Acme Corporation?", session))
print(ask("And how does that compare to Q3?", session))  # context preserved