Serverless Deployment
Deploy agents as serverless functions with handle_serverless(), Lambda adapters, and scale-to-zero.
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) orhandler()(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 OpenAIKeyGoogle 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
| Platform | Cold Start | Max Timeout | Notes |
|---|---|---|---|
| AWS Lambda | 1-3s (Python), under 1s (Go) | 15 min | Best for heavy compute |
| Cloud Functions | 1-2s | 9 min (gen1), 60 min (gen2) | GCP native |
| Cloud Run | 1-3s | 60 min | Container-based, best flexibility |
| Vercel | under 500ms | 5 min (Pro) | Best for edge/API routes |
| Azure Functions | 1-3s | 10 min | Azure 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
| Optimization | Impact | Applies To |
|---|---|---|
| Use Go SDK | 10-50x faster cold start | All platforms |
Use smaller models (gpt-4o-mini) | Faster response, lower cost | All |
| Increase memory allocation | Faster CPU = faster startup | Lambda, Cloud Functions |
| Use provisioned concurrency | Zero cold starts | Lambda (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 AMHybrid 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 demandEnvironment variables
Configure serverless agents via environment variables -- no secrets in code.
| Variable | Required | Description |
|---|---|---|
AGENTFIELD_SERVER | Yes | Control plane URL for registration and execution |
OPENAI_API_KEY | Depends | OpenAI API key (if using OpenAI models) |
ANTHROPIC_API_KEY | Depends | Anthropic API key (if using Claude models) |
FAL_KEY | No | fal.ai API key (if using media generation) |
AGENTFIELD_LOG_LEVEL | No | Logging level: DEBUG, INFO, WARN, ERROR |