Listen to this Post
Introduction
The rapid evolution of Large Language Models (LLMs) has given rise to a new paradigm in automation: AI agents that can reason, plan, and execute tasks by calling external tools. LangChain has emerged as the de facto standard framework for orchestrating these agentic workflows, reducing what once required hundreds of lines of complex orchestration code into a mere handful of API calls. Understanding how to leverage LangChain’s `create_agent` function is now essential for developers, security engineers, and AI practitioners who need to deploy intelligent automation rapidly while maintaining control over system behavior and security boundaries.
Learning Objectives
- Understand the architecture and operational mechanics of LangChain’s agent framework, including the loop-based reasoning and tool-calling paradigm
- Master the implementation of custom tools, middleware integration, and error handling patterns for production-grade AI agents
- Learn to secure agentic systems through proper input validation, tool sandboxing, and API key management across multi-provider LLM deployments
You Should Know
- Setting Up Your LangChain Environment for Agent Development
Before diving into agent creation, you need to establish a proper Python environment with all necessary dependencies. LangChain supports multiple LLM providers, and choosing the right provider involves understanding API costs, latency, and capability tradeoffs.
Linux/macOS Setup:
Create a virtual environment python3 -m venv langchain-agent source langchain-agent/bin/activate Install core dependencies pip install langchain langchain-anthropic langchain-openai python-dotenv For additional tools pip install duckduckgo-search wikipedia-api requests
Windows Setup (PowerShell):
Create a virtual environment python -m venv langchain-agent .\langchain-agent\Scripts\Activate.ps1 Install dependencies pip install langchain langchain-anthropic langchain-openai python-dotenv pip install duckduckgo-search wikipedia-api requests
Environment Configuration:
Create a `.env` file in your project root to securely store API keys:
ANTHROPIC_API_KEY=your_anthropic_key_here OPENAI_API_KEY=your_openai_key_here
Step-by-Step Setup Guide:
- Initialize the Project: Create a new directory and initialize a Python project with a `requirements.txt` file
- Configure API Keys: Never hardcode API keys; use environment variables or a secrets manager
- Test Connectivity: Run a simple test script to verify your LLM provider connection works
- Set Up Version Control: Use `.gitignore` to exclude `.env` and virtual environment directories
- Validate Python Version: Ensure you’re using Python 3.9 or higher for full LangChain compatibility
-
Building the Core Agent with Tools and System Prompts
The `create_agent` function is the fastest way to instantiate an agent with tool-calling capabilities. Understanding how to define tools properly—including type hints, docstrings, and error handling—is crucial for production deployments.
Complete Agent Implementation with Enhanced Tools:
import os
import json
import math
from typing import Optional
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from dotenv import load_dotenv
import requests
Load environment variables
load_dotenv()
Tool Definitions with Proper Error Handling
def calculator(expression: str) -> str:
"""
Performs mathematical calculations safely.
Args:
expression: A mathematical expression as a string (e.g., "2517")
Returns:
The calculated result as a string, or an error message.
"""
Security: restrict available functions to prevent injection
allowed_names = {
k: v for k, v in math.<strong>dict</strong>.items() if not k.startswith("_")
}
allowed_names.update({"abs": abs, "round": round})
try:
result = eval(expression, {"<strong>builtins</strong>": {}}, allowed_names)
return str(result)
except Exception as e:
return f"Error in calculation: {str(e)}"
def enhanced_search(query: str, max_results: int = 3) -> str:
"""
Searches the web for current information.
Args:
query: The search query string
max_results: Maximum number of results to return
Returns:
Formatted search results as a string
"""
try:
Using DuckDuckGo search API (free tier)
url = "https://api.duckduckgo.com/"
params = {
"q": query,
"format": "json",
"no_html": 1,
"skip_disambig": 1,
"max_results": max_results
}
response = requests.get(url, params=params, timeout=10)
data = response.json()
if data.get("Abstract"):
return f"Search results for '{query}':\n{data['Abstract']}"
results = []
if data.get("RelatedTopics"):
for topic in data["RelatedTopics"][:max_results]:
if "Text" in topic:
results.append(topic["Text"])
return f"Search results for '{query}':\n" + "\n".join(results) if results else f"No results found for '{query}'"
except Exception as e:
return f"Search error: {str(e)}"
def get_current_time() -> str:
"""Returns the current date and time in ISO format."""
from datetime import datetime
return datetime.now().isoformat()
Agent Creation
def create_math_search_agent():
"""Instantiate an agent with math, search, and time tools."""
Choose your model - switching providers is a configuration change
model = ChatAnthropic(
model="claude-sonnet-4-5-20250929",
temperature=0.1,
max_tokens=1024
)
Alternative: model = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0.1)
agent = create_agent(
model=model,
tools=[calculator, enhanced_search, get_current_time],
system_prompt="""You are a helpful assistant with access to tools.
- Use the calculator for any mathematical operations.
- Use the search tool when you need current information.
- Use get_current_time for time-sensitive queries.
- Always explain your reasoning before using a tool.
- If a tool returns an error, try an alternative approach or ask for clarification."""
)
return agent
Interactive Agent Execution
def run_agent_loop(agent, prompt: str):
"""Execute a user prompt and return the formatted response."""
try:
result = agent.invoke({
"messages": [{"role": "user", "content": prompt}]
})
Extract the final message content
final_message = result["messages"][-1]
return final_message.content if hasattr(final_message, 'content') else str(final_message)
except Exception as e:
return f"Agent execution failed: {str(e)}"
Example Usage
if <strong>name</strong> == "<strong>main</strong>":
agent = create_math_search_agent()
test_queries = [
"What is 25 17?",
"What is the current time?",
"Who is the current CEO of Microsoft and what is their background?"
]
for query in test_queries:
print(f"\nUser: {query}")
response = run_agent_loop(agent, query)
print(f"Agent: {response}")
print("-" 60)
Step-by-Step Implementation Guide:
- Define Tool Signatures: Use Python type hints and comprehensive docstrings—these inform both the LLM and developers about tool usage
- Implement Security Measures: In the calculator function, restrict `eval` to safe operations; in search functions, implement timeouts and response validation
- Configure the Model: Set appropriate temperature (0.0-0.3 for deterministic tasks, higher for creative work) and max_tokens
- Craft the System Be explicit about tool usage rules, error handling, and reasoning requirements
- Add Error Handling: Wrap agent invocation in try-catch blocks to handle API failures, rate limits, and tool exceptions gracefully
3. Enhancing Agents with Middleware and Guardrails
LangChain’s middleware system allows you to inject custom behavior into the agent’s execution pipeline. This is critical for implementing business logic, security guardrails, and observability.
Middleware Implementation for Production Agents:
from langchain.agents import AgentMiddleware
from typing import Dict, Any, List
import time
import logging
Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(<strong>name</strong>)
class LoggingMiddleware(AgentMiddleware):
"""Logs agent inputs and outputs for auditing."""
def pre_process(self, state: Dict[str, Any]) -> Dict[str, Any]:
"""Called before agent processing."""
messages = state.get("messages", [])
user_message = messages[-1]["content"] if messages else "No messages"
logger.info(f"Agent processing user input: {user_message[:100]}...")
return state
def post_process(self, state: Dict[str, Any]) -> Dict[str, Any]:
"""Called after agent processing."""
messages = state.get("messages", [])
if messages:
last_response = messages[-1].get("content", "")[:100]
logger.info(f"Agent response: {last_response}...")
return state
class RateLimitMiddleware(AgentMiddleware):
"""Implements simple rate limiting for API calls."""
def <strong>init</strong>(self, max_calls_per_minute: int = 10):
self.max_calls = max_calls_per_minute
self.call_timestamps: List[bash] = []
def pre_process(self, state: Dict[str, Any]) -> Dict[str, Any]:
current_time = time.time()
Remove timestamps older than 60 seconds
self.call_timestamps = [t for t in self.call_timestamps if current_time - t < 60]
if len(self.call_timestamps) >= self.max_calls:
raise Exception("Rate limit exceeded. Please wait and try again.")
self.call_timestamps.append(current_time)
return state
class ContentGuardrailMiddleware(AgentMiddleware):
"""Blocks prohibited topics and validates input/output."""
PROHIBITED_KEYWORDS = ["attack", "exploit", "hack", "malware", "phishing"]
def pre_process(self, state: Dict[str, Any]) -> Dict[str, Any]:
messages = state.get("messages", [])
for msg in messages:
content = msg.get("content", "").lower()
for keyword in self.PROHIBITED_KEYWORDS:
if keyword in content:
raise ValueError(f"Input contains prohibited content: {keyword}")
return state
Apply middleware to your agent
def create_secured_agent():
model = ChatAnthropic(model="claude-sonnet-4-5-20250929")
agent = create_agent(
model=model,
tools=[calculator, enhanced_search],
system_prompt="You are a helpful assistant with security guardrails.",
middleware=[
LoggingMiddleware(),
RateLimitMiddleware(max_calls_per_minute=10),
ContentGuardrailMiddleware()
]
)
return agent
Step-by-Step Middleware Implementation:
- Define Custom Middleware Classes: Extend `AgentMiddleware` and implement `pre_process` and optionally `post_process`
2. Implement Logging: Track all user inputs and agent responses for audit trails and debugging - Add Rate Limiting: Prevent API abuse by tracking call frequency using timestamps
- Deploy Content Guardrails: Scan input and output for prohibited keywords or patterns
- Test Middleware Chain: Verify middleware execution order (pre-process runs before, post-process after)
4. Scaling Up with LangGraph for Complex Workflows
While `create_agent` is ideal for straightforward agent use cases, LangGraph provides the underlying architecture for complex, multi-step workflows with state persistence, human-in-the-loop, and conditional routing.
Transitioning from create_agent to LangGraph:
from typing import List, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
class AgentState(TypedDict):
"""State for the agent workflow."""
messages: List[Dict[str, str]]
tool_results: List[bash]
current_step: str
iteration_count: int
def build_langgraph_agent():
"""Create a stateful agent using LangGraph."""
graph = StateGraph(AgentState)
def call_model(state: AgentState):
"""Node that calls the LLM."""
response = model.invoke(state["messages"])
return {
"messages": state["messages"] + [{"role": "assistant", "content": response.content}],
"current_step": "processing"
}
def call_tools(state: AgentState):
"""Node that executes tools."""
Tool execution logic here
return {
"tool_results": state["tool_results"] + ["Tool executed"],
"current_step": "tools"
}
def should_continue(state: AgentState) -> str:
"""Conditional routing logic."""
if state.get("iteration_count", 0) >= 3:
return "end"
return "call_model"
Build the graph
graph.add_node("call_model", call_model)
graph.add_node("call_tools", call_tools)
graph.set_entry_point("call_model")
graph.add_conditional_edges("call_model", should_continue, {
"call_model": "call_model",
"call_tools": "call_tools",
"end": END
})
graph.add_edge("call_tools", "call_model")
memory = MemorySaver()
return graph.compile(checkpointer=memory)
Use the stateful agent
agent = build_langgraph_agent()
config = {"configurable": {"thread_id": "session_001"}}
result = agent.invoke({"messages": [{"role": "user", "content": "Help me solve 2517"}]}, config)
Step-by-Step LangGraph Migration:
- Define State Schema: Create a TypedDict that holds all state variables (messages, tool results, iteration count)
- Build Nodes: Create functions for each processing step (model call, tool execution, validation)
- Add Edges: Define entry point and conditional routing between nodes
- Implement Checkpointing: Use MemorySaver or database-backed checkpoints for state persistence
- Configure Threading: Use thread IDs to maintain separate conversation sessions
5. Agentic Security: Hardening Multi-Provider Deployments
Security is paramount when deploying AI agents that interact with external systems and sensitive data. A comprehensive security approach covers API key management, tool sandboxing, input validation, and output filtering.
API Key Management with Vault Integration:
import hvac HashiCorp Vault client
import os
class VaultKeyManager:
"""Secure API key management using HashiCorp Vault."""
def <strong>init</strong>(self, vault_addr: str, vault_token: str):
self.client = hvac.Client(url=vault_addr, token=vault_token)
def get_key(self, key_path: str) -> Optional[bash]:
"""Retrieve a secret from Vault."""
try:
response = self.client.secrets.kv.v2.read_secret_version(path=key_path)
return response['data']['data'].get('api_key')
except Exception as e:
logger.error(f"Failed to retrieve key from Vault: {e}")
return None
Usage
vault = VaultKeyManager(
vault_addr=os.getenv("VAULT_ADDR", "http://localhost:8200"),
vault_token=os.getenv("VAULT_TOKEN", "")
)
api_key = vault.get_key("anthropic-production")
Tool Sandboxing for Security:
import subprocess
import resource
import signal
import tempfile
def sandboxed_tool_execution(script: str, timeout: int = 5):
"""Execute a script in a sandboxed environment with resource limits."""
Create a temporary directory for execution
with tempfile.TemporaryDirectory() as tmpdir:
script_path = os.path.join(tmpdir, "script.py")
with open(script_path, "w") as f:
f.write(script)
Set resource limits
resource.setrlimit(resource.RLIMIT_CPU, (timeout, timeout + 1))
resource.setrlimit(resource.RLIMIT_AS, (1024 1024 50, 1024 1024 50)) 50MB
try:
result = subprocess.run(
["python3", script_path],
capture_output=True,
text=True,
timeout=timeout,
cwd=tmpdir
)
return result.stdout if result.returncode == 0 else result.stderr
except subprocess.TimeoutExpired:
return "Execution timeout - sandbox prevented infinite loop"
except Exception as e:
return f"Sandbox error: {str(e)}"
Security Hardening Checklist for Agentic Systems:
- API Key Rotation: Implement automatic rotation of API keys using Vault or AWS Secrets Manager
- Input Validation: Sanitize all user inputs to prevent prompt injection attacks
- Tool Isolation: Run external tools in isolated containers or sandboxes
- Output Filtering: Scan agent outputs for sensitive data patterns (PII, API keys, passwords)
- Audit Logging: Maintain comprehensive logs of all agent interactions for security reviews
- Rate Limiting: Implement exponential backoff and retry mechanisms
- Model Hardening: Use system prompts to instruct models not to share internal reasoning or sensitive details
6. Integration with Enterprise Systems and APIs
Real-world agents need to interact with internal APIs, databases, and enterprise systems. This section covers secure integration patterns.
Secure API Tool with Authentication:
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class SecureAPIClient:
"""Secure API client with retry logic and token management."""
def <strong>init</strong>(self, base_url: str, token_url: str, client_id: str, client_secret: str):
self.base_url = base_url
self.token_url = token_url
self.client_id = client_id
self.client_secret = client_secret
self.access_token = None
self.token_expiry = 0
Setup session with retry
self.session = requests.Session()
retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
def get_token(self) -> str:
"""Obtain OAuth2 token with client credentials flow."""
if self.access_token and time.time() < self.token_expiry:
return self.access_token
response = self.session.post(
self.token_url,
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
)
response.raise_for_status()
token_data = response.json()
self.access_token = token_data['access_token']
self.token_expiry = time.time() + token_data['expires_in'] - 60
return self.access_token
def secure_call(self, endpoint: str, method: str = "GET", data: dict = None):
"""Make an authenticated API call."""
token = self.get_token()
headers = {"Authorization": f"Bearer {token}"}
url = f"{self.base_url}/{endpoint.lstrip('/')}"
if method.upper() == "GET":
response = self.session.get(url, headers=headers, params=data)
elif method.upper() == "POST":
response = self.session.post(url, headers=headers, json=data)
else:
raise ValueError(f"Unsupported method: {method}")
response.raise_for_status()
return response.json()
Tool for LangChain agent
def query_crm(customer_id: str) -> str:
"""Query the internal CRM system for customer information."""
client = SecureAPIClient(
base_url=os.getenv("CRM_BASE_URL"),
token_url=os.getenv("CRM_TOKEN_URL"),
client_id=os.getenv("CRM_CLIENT_ID"),
client_secret=os.getenv("CRM_CLIENT_SECRET")
)
try:
result = client.secure_call(f"/customers/{customer_id}")
return json.dumps(result, indent=2)
except Exception as e:
return f"CRM query failed: {str(e)}"
Step-by-Step Enterprise Integration:
- Implement Token-Based Authentication: Use OAuth2 or API keys with proper rotation
- Add Retry Logic: Handle transient failures with exponential backoff
- Implement Circuit Breaker: Prevent cascading failures when external services are unavailable
- Validate Responses: Check API responses for expected structure before passing to agent
- Log All External Calls: Maintain audit trails for compliance and debugging
What Undercode Say
- Key Takeaway 1: LangChain’s `create_agent` reduces agent development from weeks to minutes, but production-ready systems require careful attention to middleware, security, and error handling. The five-line example is a great starting point, but real-world deployments need the additional layers covered in this guide.
-
Key Takeaway 2: The true power of LangChain lies in its extensibility—from simple tool calling with `create_agent` to complex stateful workflows with LangGraph. The middleware system provides enterprise-grade capabilities like auditing, rate limiting, and guardrails without sacrificing the simplicity that makes LangChain accessible to newcomers.
Analysis: The LangChain ecosystem has matured significantly over the past year. What began as a simple orchestration library now provides a complete stack for building production AI agents. The `create_agent` function represents a sweet spot between simplicity and control, allowing developers to start quickly while providing clear migration paths to more sophisticated patterns. Security remains a critical concern; however, the framework’s middleware architecture and integration with tools like Vault enable robust security implementations. The multi-provider support is particularly valuable for enterprises needing redundancy and cost optimization across LLM providers. As organizations increasingly adopt agentic AI, patterns like tool sandboxing, state management, and content guardrails will become standard practice. The shift from simple agents to complex multi-agent systems is already underway, and LangGraph’s graph-based architecture positions LangChain well for this evolution.
Prediction
+1 The commoditization of AI agent development through frameworks like LangChain will accelerate enterprise adoption of intelligent automation, with a projected 40% increase in production agent deployments by 2027.
+1 Standardization around agentic patterns will emerge, with LangChain’s architecture becoming a reference implementation for tool-calling systems, similar to how React standardized frontend component architectures.
-1 Security vulnerabilities in agentic systems—particularly prompt injection and tool misuse—will become a significant attack vector, requiring new security paradigms and specialized security tools.
+1 The integration of LangChain with enterprise systems through secure API patterns will create new opportunities for AI-driven business process automation, particularly in customer service, data analysis, and workflow orchestration.
-1 As agent complexity increases, debugging and observability will become challenging, necessitating new tools for tracing agent reasoning and tool execution—an area where LangChain currently has limited built-in capabilities.
▶️ Related Video (80% Match):
🎯Let’s Practice For Free:
🎓 Live Courses & Certifications:
Join Undercode Academy for Verified Certifications
🚀 Request a Custom Project:
Secure, high-velocity infrastructure and disruptive technological engineering. Contact our engineering team for high-tier development and proprietary systems:
[email protected]
💎 Smart Architecture | 🛡️ Secure by Design | ⭐ Trusted by Thousands
IT/Security Reporter URL:
Reported By: Veera Malla – Hackers Feeds
Extra Hub: Undercode MoN
Basic Verification: Pass ✅
