AgentFieldreference

Serverless Deployment

Deploy agents as serverless functions with handle_serverless(), Lambda adapters, and scale-to-zero.

Serverless deployment — same Agent, serverless runtime

Deploy agents as Lambda functions, Cloud Functions, or edge handlers while keeping the same agent logic.

Not every agent needs a long-running server. For event-driven workloads, batch processing, or cost-sensitive deployments, AgentField agents can run as serverless functions. Python's handle_serverless(event), TypeScript's handler(), and Go's Handler() wrap your agent so it works with AWS Lambda, Google Cloud Functions, Azure Functions, Vercel, and any platform that speaks HTTP.

from agentfield import Agent, AIConfig
from pydantic import BaseModel

app = Agent(
    node_id="invoice-processor",
    ai_config=AIConfig(model="openai/gpt-4o-mini"),
)

class InvoiceInput(BaseModel):
    document_url: str
    vendor: str

class InvoiceOutput(BaseModel):
    total: float
    line_items: list[dict]
    due_date: str
    confidence: float

@app.reasoner()
async def process_invoice(document_url: str, vendor: str) -> dict:
    result = await app.ai(
        system="Extract structured invoice data.",
        user=f"Vendor: {vendor}, Document: {document_url}",
        schema=InvoiceOutput,
    )
    return result

# Lambda handler — pass the event directly
def handler(event, context):
    return app.handle_serverless(event)

What just happened

The same agent definition was exposed as a serverless handler without rewriting the business logic. You keep the same reasoners and schemas, then swap only the runtime boundary.

{
  "deployment_shape": "serverless",
  "entrypoint": "handle_serverless / handler",
  "cold_start_behavior": "runtime-dependent",
  "scale_model": "scale_to_zero"
}
What you get
  • Serverless adapter -- handle_serverless() (Python) or handler() (TypeScript/Go) wraps your agent for any FaaS platform.
  • Registration behavior -- startup and registration behavior depend on your runtime and deployment adapter.
  • Scale-to-zero -- no traffic, no cost. The control plane routes requests when the function wakes up.
  • Typed I/O -- Pydantic models, Zod schemas, or Go structs for request/response validation.
  • Platform support -- Lambda, Cloud Functions, Cloud Run, Vercel, Netlify, Azure Functions.
  • Cold start optimization -- use fast models and the Go SDK for minimal startup latency.
Platform adapters

Each adapter translates the platform's event format into AgentField's execution model.

AWS Lambda

# handler.py — deploy with SAM, CDK, or Serverless Framework
from agentfield import Agent

app = Agent(node_id="my-agent")

@app.reasoner()
async def process(text: str) -> dict:
    return {"result": await app.ai(user=text)}

def handler(event, context):
    return app.handle_serverless(event)
# template.yaml (AWS SAM)
Resources:
  AgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: handler.handler
      Runtime: python3.12
      Timeout: 300
      MemorySize: 512
      Environment:
        Variables:
          AGENTFIELD_SERVER: !Ref ControlPlaneUrl
          OPENAI_API_KEY: !Ref OpenAIKey

Google Cloud Functions

# main.py — deploy with gcloud functions deploy
from agentfield import Agent

app = Agent(node_id="my-agent")

@app.reasoner()
async def process(text: str) -> dict:
    return {"result": await app.ai(user=text)}

# Cloud Functions entry point — pass the event directly
def handle(event, context=None):
    return app.handle_serverless(event)

Vercel / Netlify Edge

// api/agent.ts — auto-detected by Vercel
import { Agent } from "@agentfield/sdk";

const agent = new Agent({ nodeId: "my-agent" });

agent.reasoner("process", async (ctx) => {
  return { result: await ctx.ai(JSON.stringify(ctx.input)) };
});

export default agent.handler();

Platform Comparison

PlatformCold StartMax TimeoutNotes
AWS Lambda1-3s (Python), under 1s (Go)15 minBest for heavy compute
Cloud Functions1-2s9 min (gen1), 60 min (gen2)GCP native
Cloud Run1-3s60 minContainer-based, best flexibility
Vercelunder 500ms5 min (Pro)Best for edge/API routes
Azure Functions1-3s10 minAzure ecosystem
Cold start optimization

Serverless cold starts can add latency. These patterns minimize startup time.

Use a Fast Model

app = Agent(
    node_id="my-agent",
    ai_config=AIConfig(
        model="openai/gpt-4o-mini",      # fast, cheap model for serverless
    ),
)

Tips

OptimizationImpactApplies To
Use Go SDK10-50x faster cold startAll platforms
Use smaller models (gpt-4o-mini)Faster response, lower costAll
Increase memory allocationFaster CPU = faster startupLambda, Cloud Functions
Use provisioned concurrencyZero cold startsLambda (costs more)
Patterns

Event-Driven Processing

Trigger agent execution from cloud events (S3 uploads, Pub/Sub messages, webhooks).

from agentfield import Agent

app = Agent(node_id="document-processor")

@app.reasoner()
async def process_upload(bucket: str, key: str) -> dict:
    # Download and process the uploaded document
    analysis = await app.ai(
        system="Analyze this document and extract key information.",
        user=f"Document from s3://{bucket}/{key}",
    )
    return {"analysis": analysis, "source": f"s3://{bucket}/{key}"}

def handler(event, context):
    # Normalize S3 event into agent input
    record = event["Records"][0]["s3"]
    normalized = {
        "reasoner": "process_upload",
        "input": {
            "bucket": record["bucket"]["name"],
            "key": record["object"]["key"],
        },
    }
    return app.handle_serverless(normalized)

Scheduled Agent Runs

Run agents on a cron schedule with CloudWatch Events or Cloud Scheduler.

@app.reasoner()
async def daily_report(period: str = "24h") -> dict:
    # Fetch data from other agents
    metrics = await app.call("metrics.collect", period=period)
    anomalies = await app.call("anomaly-detector.scan", **metrics)

    report = await app.ai(
        system="Generate a daily operations report.",
        user=str({"metrics": metrics, "anomalies": anomalies}),
    )
    return {"report": report, "anomaly_count": len(anomalies.get("items", []))}

def handler(event, context):
    return app.handle_serverless(event)
# Trigger with: cron(0 9 * * ? *) — every day at 9 AM

Hybrid Deployment

Run critical agents on servers, burst agents as serverless.

import asyncio
from agentfield import Agent

# Long-running orchestrator on a server
orchestrator = Agent(node_id="orchestrator")

@orchestrator.reasoner()
async def coordinate(task: str) -> dict:
    # Fan out to serverless workers — they scale automatically
    results = await asyncio.gather(
        orchestrator.call("serverless-analyzer.process", chunk=1),
        orchestrator.call("serverless-analyzer.process", chunk=2),
        orchestrator.call("serverless-analyzer.process", chunk=3),
    )
    return {"combined": results}

orchestrator.serve()  # Long-running server

# The "serverless-analyzer" agents are deployed as Lambda functions
# and scale independently based on demand
Environment variables

Configure serverless agents via environment variables -- no secrets in code.

VariableRequiredDescription
AGENTFIELD_SERVERYesControl plane URL for registration and execution
OPENAI_API_KEYDependsOpenAI API key (if using OpenAI models)
ANTHROPIC_API_KEYDependsAnthropic API key (if using Claude models)
FAL_KEYNofal.ai API key (if using media generation)
AGENTFIELD_LOG_LEVELNoLogging level: DEBUG, INFO, WARN, ERROR