Most developers new to AI agents encounter the same problem: the frameworks designed to orchestrate LLM workflows introduce additional layers and features that may not be necessary for what they actually need. LangChain, AutoGen, and similar tools are powerful, but they add complexity, dependencies, and overhead that can slow down prototyping and inflate cloud costs. If you’re building a focused agent for a specific task, you don’t always need that machinery.
The good news is that the Anthropic SDK makes it straightforward to build production-ready agents in pure Python. You get direct control over the agent loop, tool calling, state management, and streaming responses without abstraction layers getting in the way. This is especially valuable for serverless deployments, rapid prototypes, and teams that prefer readable, maintainable code over framework magic.
Here’s how to build this, from a simple agent to patterns you can use in real applications.
Why Lightweight Agents Matter
Framework-heavy approaches excel at orchestrating complex multi-agent workflows with memory persistence, dynamic tool loading, and enterprise monitoring. They’re built for scale and abstraction.
But if you’re building a single-purpose agent, you’re paying for features you don’t use. Lightweight agents give you:
- Direct control over the agent loop and decision logic
- Minimal dependencies and faster cold starts on serverless platforms
- Predictable costs because you know exactly what you’re calling
- Easier debugging and testing since the code is straightforward
- Freedom to design state management exactly how you need it
The tradeoff is that you’re responsible for handling the agent loop, tool management, and state yourself. For many use cases, that’s a fair deal.
The Agent Loop: The Core Pattern
An AI agent is fundamentally a loop: the model sees the current state, decides what to do next (call a tool or respond), and the loop continues until the task is complete. Here’s the bare minimum structure:
import anthropic
def run_agent(user_message: str) -> str:
client = anthropic.Anthropic()
messages = [{"role": "user", "content": user_message}]
while True:
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=TOOLS, # Define tools here
messages=messages
)
# Check if the model wants to use a tool
if response.stop_reason == "tool_use":
# Find the tool use block
tool_use = next(
(block for block in response.content if block.type == "tool_use"),
None
)
if tool_use:
# Execute the tool
result = execute_tool(tool_use.name, tool_use.input)
# Add the assistant response and tool result to messages
messages.append({"role": "assistant", "content": response.content})
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(result)
}]
})
else:
# Model has finished, extract text response
final_response = next(
(block.text for block in response.content if hasattr(block, "text")),
None
)
return final_response
This pattern handles the fundamental agent behavior: ask the model, check if it needs tools, execute tools, and loop until you get a final response. No framework, no magic. Just clear, testable code.
Defining and Using Tools
Tools are how agents interact with the outside world. Define them as JSON schemas that describe what the tool does and what parameters it accepts:
TOOLS = [
{
"name": "search_documents",
"description": "Search through documents for relevant information",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"limit": {
"type": "integer",
"description": "Maximum number of results",
"default": 5
}
},
"required": ["query"]
}
},
{
"name": "fetch_document",
"description": "Retrieve the full content of a specific document",
"input_schema": {
"type": "object",
"properties": {
"doc_id": {
"type": "string",
"description": "The unique identifier of the document"
}
},
"required": ["doc_id"]
}
}
]
def execute_tool(name: str, input_data: dict) -> str:
"""Execute a tool and return the result as a string"""
if name == "search_documents":
query = input_data.get("query")
limit = input_data.get("limit", 5)
# Your actual search logic here
return f"Found documents matching '{query}' (limited to {limit})"
elif name == "fetch_document":
doc_id = input_data.get("doc_id")
# Your actual document fetch logic here
return f"Document content for {doc_id}..."
return "Tool not found"
The key is that each tool definition tells Claude exactly what it can do and what parameters to use. Claude will generate the tool calls, and you execute them and feed the results back.
Streaming Responses for Better UX
For interactive applications, streaming the model’s response as it’s being generated gives users immediate feedback. The Anthropic SDK makes this straightforward:
def run_agent_with_streaming(user_message: str):
client = anthropic.Anthropic()
messages = [{"role": "user", "content": user_message}]
while True:
with client.messages.stream(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=TOOLS,
messages=messages
) as stream:
# Collect the full response
full_response = stream.get_final_message()
if full_response.stop_reason == "tool_use":
tool_use = next(
(block for block in full_response.content if block.type == "tool_use"),
None
)
if tool_use:
result = execute_tool(tool_use.name, tool_use.input)
messages.append({"role": "assistant", "content": full_response.content})
messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(result)
}]
})
else:
# Stream the final text response to the user
for text in stream.text_stream:
print(text, end="", flush=True)
print() # Newline after streaming
break
This approach uses the streaming API while still collecting the full message object so you can check for tool use. Users see responses appear in real time, which feels more responsive than waiting for the entire response.
Real-World Pattern: A Document Analyzer Agent
Here’s a concrete example. A document analyzer agent that can search, fetch, and summarize documents:
class DocumentAnalyzerAgent:
def __init__(self, documents: dict):
"""Initialize with a collection of documents."""
self.client = anthropic.Anthropic()
self.documents = documents # {doc_id: content}
self.conversation_history = []
def search_documents(self, query: str, limit: int = 5) -> str:
"""Simple keyword search across documents."""
results = []
query_lower = query.lower()
for doc_id, content in self.documents.items():
if query_lower in content.lower():
results.append(doc_id)
if len(results) >= limit:
break
return f"Found {len(results)} documents: {', '.join(results)}"
def fetch_document(self, doc_id: str) -> str:
"""Retrieve full document content."""
return self.documents.get(doc_id, "Document not found")
def execute_tool(self, name: str, input_data: dict) -> str:
"""Route tool calls to the appropriate handler."""
if name == "search_documents":
return self.search_documents(
input_data.get("query"),
input_data.get("limit", 5)
)
elif name == "fetch_document":
return self.fetch_document(input_data.get("doc_id"))
return "Unknown tool"
def analyze(self, user_query: str) -> str:
"""Run the agent loop with the user query."""
self.conversation_history.append({"role": "user", "content": user_query})
system_prompt = """You are a helpful document analyst. Help the user understand
and analyze documents. Use the search_documents tool to find relevant documents,
then fetch_document to read their content. Provide clear, concise summaries."""
tools = [
{
"name": "search_documents",
"description": "Search documents by keyword",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "integer", "default": 5}
},
"required": ["query"]
}
},
{
"name": "fetch_document",
"description": "Get the full content of a document",
"input_schema": {
"type": "object",
"properties": {
"doc_id": {"type": "string"}
},
"required": ["doc_id"]
}
}
]
max_iterations = 10
iteration = 0
while iteration < max_iterations:
iteration += 1
response = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=system_prompt,
tools=tools,
messages=self.conversation_history
)
if response.stop_reason == "tool_use":
# Extract and execute tool calls
tool_uses = [
block for block in response.content
if block.type == "tool_use"
]
# Add assistant response to history
self.conversation_history.append({
"role": "assistant",
"content": response.content
})
# Execute tools and add results
tool_results = []
for tool_use in tool_uses:
result = self.execute_tool(tool_use.name, tool_use.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": result
})
self.conversation_history.append({
"role": "user",
"content": tool_results
})
else:
# Extract final text response
final_text = next(
(block.text for block in response.content if hasattr(block, "text")),
None
)
if final_text:
self.conversation_history.append({
"role": "assistant",
"content": final_text
})
return final_text
return "Agent reached maximum iterations"
# Usage example
if __name__ == "__main__":
docs = {
"doc_1": "Python is a high-level programming language known for simplicity.",
"doc_2": "Anthropic builds AI systems that are safe and interpretable.",
"doc_3": "Serverless computing reduces operational overhead significantly."
}
agent = DocumentAnalyzerAgent(docs)
result = agent.analyze("What do the documents say about Python and AI?")
print(result)
This agent maintains conversation history, handles multiple tool calls in a single response, and provides a clean interface for analysis tasks. It’s ready to extend with more tools and deploy to production environments.
State Management and Persistence
For agents that need to remember state across requests, you have several options depending on your deployment model:
In-Memory State (Single Process)
If your agent runs in a single long-lived process, keep conversation history as an instance variable. This is simple but doesn’t survive restarts.
Serialized State (Serverless)
For serverless functions, serialize the conversation history to a database or cache after each invocation:
import json
import boto3
def save_agent_state(agent_id: str, messages: list):
"""Save agent conversation history to DynamoDB."""
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("agent_state")
table.put_item(
Item={
"agent_id": agent_id,
"messages": json.dumps(messages),
"timestamp": int(time.time())
}
)
def load_agent_state(agent_id: str) -> list:
"""Load agent conversation history from DynamoDB."""
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("agent_state")
response = table.get_item(Key={"agent_id": agent_id})
if "Item" in response:
return json.loads(response["Item"]["messages"])
return []
def lambda_handler(event, context):
agent_id = event.get("agent_id")
user_message = event.get("message")
# Load previous conversation
messages = load_agent_state(agent_id)
messages.append({"role": "user", "content": user_message})
# Run agent (simplified)
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=messages
)
# Extract response and save state
final_text = next(
(block.text for block in response.content if hasattr(block, "text")),
None
)
messages.append({"role": "assistant", "content": final_text})
save_agent_state(agent_id, messages)
return {"statusCode": 200, "body": final_text}
This pattern works well for serverless platforms like AWS Lambda, where each invocation is stateless. Store just the messages, not the entire agent object, to keep serialization simple.
Cost Considerations and Optimization
One advantage of lightweight agents is that you control exactly what you send to the API. Here are practical ways to keep costs down:
Limit Conversation History
Don’t send the entire conversation to every request. Summarize old messages or use a sliding window:
def trim_conversation_history(messages: list, max_messages: int = 20) -> list:
"""Keep only the most recent messages to reduce token usage."""
if len(messages) <= max_messages:
return messages
# Keep system message (if present) and recent messages
return messages[:1] + messages[-(max_messages - 1):]
Use Caching for Repeated Queries
If your agent performs similar analyses frequently, cache tool results:
from functools import lru_cache
@lru_cache(maxsize=128)
def cached_search(query: str, limit: int) -> str:
"""Cache search results to avoid redundant queries."""
# Your search logic here
return results
Choose the Right Model
Claude 3.5 Sonnet offers a good balance of capability and cost for most agent tasks. For simpler tasks, consider Haiku. For complex reasoning, Opus. Match the model to the job.
Batch Process When Possible
If you’re processing many documents or queries, use the Anthropic Batch API to reduce per-token costs by 50%.
Lightweight vs. Framework-Heavy: When to Choose Each
Choose Lightweight (Python + Anthropic SDK) When:
- You’re building a single-purpose agent with a clear task
- You need fast cold starts for serverless deployments
- Your team prefers readable, maintainable code over abstraction
- You want fine-grained control over the agent loop and tool management
- Cost per inference matters and you want to minimize dependencies
Choose a Framework (LangChain, AutoGen) When:
- You’re orchestrating multiple agents or complex workflows
- You need built-in memory management, retrieval, and persistence
- Your team values the ecosystem and integrations a framework provides
- You’re building long-lived applications with evolving requirements
- Enterprise monitoring and observability are requirements
The lightweight approach isn’t about frameworks being bad. It’s about choosing the right tool for the job. A simple agent doesn’t need a framework any more than a simple script needs a full application framework.
Deploying Lightweight Agents to Serverless Platforms
The simplicity of a lightweight agent makes it ideal for serverless. Here’s a quick example for AWS Lambda:
import anthropic
import json
def lambda_handler(event, context):
"""Handle incoming requests from API Gateway."""
try:
body = json.loads(event.get("body", "{}"))
user_message = body.get("message")
if not user_message:
return {
"statusCode": 400,
"body": json.dumps({"error": "Missing message"})
}
# Initialize agent
client = anthropic.Anthropic()
messages = [{"role": "user", "content": user_message}]
# Run agent loop (simplified for brevity)
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=messages
)
result = next(
(block.text for block in response.content if hasattr(block, "text")),
None
)
return {
"statusCode": 200,
"body": json.dumps({"response": result})
}
except Exception as e:
return {
"statusCode": 500,
"body": json.dumps({"error": str(e)})
}
Package your agent code with the Anthropic SDK, deploy to Lambda, and you have a production API. The entire package is lightweight enough for fast cold starts and low memory usage.
Wrapping Up
Building AI agents doesn’t require heavy frameworks. Python and the Anthropic SDK give you everything you need to build focused, production-ready agents that are easy to understand, debug, and deploy.
The pattern is simple: define your tools, run the agent loop, handle tool calls, and stream responses. From there, you can add persistence, optimize costs, and scale to serverless platforms without introducing unnecessary complexity.
Start lightweight. Add complexity only when you actually need it. That’s how you build systems that work well and stay maintainable.
Can I use multiple tools in a single agent turn?
Yes. When the model’s stop_reason is “tool_use”, there may be multiple tool_use blocks in response.content. Collect all of them, execute each one, and send all results back in a single tool_result message. This is more efficient than making separate API calls.
How do I handle tool execution errors?
Catch exceptions in your execute_tool function and return the error as a string in the tool_result. Claude will see the error and can adjust its approach, retry with different parameters, or inform the user. Always return a string, even for errors.
What’s the difference between streaming and non-streaming responses?
Streaming returns tokens as they’re generated, giving users immediate feedback. Non-streaming waits for the full response. Both work with tool use. Streaming is better for interactive applications; non-streaming is simpler for batch processing.
Should I use conversation history in serverless functions?
Yes, but store it externally (database, cache, or file storage). Each Lambda invocation is stateless. Load the history at the start, add the new message, run the agent, and save the updated history before returning. This lets you maintain multi-turn conversations across invocations.
How do I know if my agent is looping infinitely?
Always set a max_iterations limit in your agent loop and break out if you reach it. Also monitor token usage; if a single request is using unusually high tokens, the agent may be stuck. Test with simple tasks first to ensure the loop terminates correctly.
