Securely Query Postgres from LangGraph Using Arcade MCP

Building AI agents that interact with databases presents significant technical challenges. Authentication, connection management, and secure query execution often become roadblocks that prevent agents from reaching production. This guide shows you how to leverage Arcade's Model Context Protocol (MCP) support to connect LangGraph agents with Postgres databases, enabling secure database operations without managing advanced infrastructure.

The Database Integration Challenge for AI Agents

Traditional approaches to database connectivity for AI agents require managing connection strings, handling authentication, and implementing security boundaries—all while ensuring the LLM never sees sensitive credentials. Less than 30% of AI projects reach production because agents cannot obtain secure, user-scoped credentials to the systems they must act on.

Arcade.dev bridges critical gaps in MCP by providing authentication-first architecture with secure, OAuth-based connections that let AI access tools as the end user, making database operations both secure and scalable.

MCP Architecture for Database Operations

Model Context Protocol represents a significant step toward standardizing how AI agents interact with tools and external systems. When it comes to database operations, MCP provides several key advantages:

Standardized tool definitions: Database operations become LLM-consumable tools with proper context
Security isolation: Credentials remain separate from the AI model
Multi-user support: Each user's database access remains isolated
Production readiness: Built-in monitoring, logging, and evaluation capabilities

MCP's streamable HTTP transport creates an open standard for connecting AI tools, and Arcade's native support makes this even more powerful for developers. This flexibility allows you to combine database tools with other integrations seamlessly.

Setting Up Arcade with Postgres MCP Server

Prerequisites

Before starting, ensure you have:

An Arcade.dev account with API key
Python 3.8+ or Node.js 18+ installed
A Postgres database with connection credentials
LangGraph installed in your environment

Installing the Arcade Engine

The Arcade Engine serves as your MCP server and tool provider. It allows you to write your tools once and serve them across any LLM or orchestration framework, no matter the protocol.

For Local Development

Install the Arcade CLI and engine:

# Install Arcade CLI
pip install arcade-ai

# For macOS
brew tap arcadeai/arcade
brew install arcade-engine

# For Ubuntu/Debian
curl -fsSL https://download.arcade.dev/arcade.gpg | sudo gpg --dearmor -o /usr/share/keyrings/arcade.gpg
echo "deb [signed-by=/usr/share/keyrings/arcade.gpg] https://download.arcade.dev/apt stable main" | sudo tee /etc/apt/sources.list.d/arcade.list
sudo apt update && sudo apt install arcade-engine

Configuring the Postgres MCP Server

Arcade's registry includes official MCP servers for databases including Postgres. Configure your Postgres connection in the engine configuration:

# engine.yaml
auth:
  providers:
    - id: postgres-provider
      description: "Postgres database connection"
      enabled: true
      type: database
      provider_id: postgres

workers:
  - id: "postgres-mcp-worker"
    enabled: true
    mcp:
      server_name: "postgres"
      transport: http
      config:
        host: ${env:POSTGRES_HOST}
        port: ${env:POSTGRES_PORT}
        database: ${env:POSTGRES_DATABASE}
        user: ${env:POSTGRES_USER}
        password: ${env:POSTGRES_PASSWORD}
        ssl: true
        max_connections: 10

Set your environment variables:

export ARCADE_API_KEY="your_arcade_api_key"
export POSTGRES_HOST="your_postgres_host"
export POSTGRES_PORT="5432"
export POSTGRES_DATABASE="your_database"
export POSTGRES_USER="your_username"
export POSTGRES_PASSWORD="your_password"

Integrating Postgres Tools with LangGraph

Python Implementation

Here's how to integrate Postgres database operations into your LangGraph application using Arcade:

import os
from langchain_arcade import ArcadeToolManager
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent

# Initialize Arcade Tool Manager
arcade_api_key = os.environ["ARCADE_API_KEY"]
tool_manager = ArcadeToolManager(api_key=arcade_api_key)

# Get Postgres tools from the MCP server
postgres_tools = tool_manager.get_tools(toolkits=["Postgres"])

# Available Postgres tools include:
# - Postgres.QueryDatabase: Execute read queries
# - Postgres.GetSchema: Retrieve database schema
# - Postgres.ListTables: List all tables
# - Postgres.DescribeTable: Get table structure

# Create language model with tools
model = ChatOpenAI(model="gpt-4o")
bound_model = model.bind_tools(postgres_tools)

# Initialize memory for conversation state
memory = MemorySaver()

# Create ReAct agent
graph = create_react_agent(
    model=bound_model,
    tools=postgres_tools,
    checkpointer=memory
)

# Configuration with user context
config = {
    "configurable": {
        "thread_id": "postgres_session_1",
        "user_id": "user@example.com"  # User-specific database access
    }
}

# Query the database
user_query = {
    "messages": [
        ("user", "Show me the top 10 customers by total order value from our database")
    ]
}

# Stream the response
for chunk in graph.stream(user_query, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

JavaScript/TypeScript Implementation

For JavaScript developers using LangGraph:

import { Arcade } from "@arcadeai/arcadejs";
import { toZod } from "@arcadeai/arcadejs/lib";
import { tool } from "@langchain/core/tools";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { ChatOpenAI } from "@langchain/openai";
import { MemorySaver } from "@langchain/langgraph";

// Initialize Arcade client
const arcade = new Arcade({
    apiKey: process.env.ARCADE_API_KEY
});

// Get Postgres tools from MCP server
const postgresToolkit = await arcade.tools.list({
    toolkit: "Postgres",
    limit: 30
});

// Convert to LangGraph-compatible tools
const arcadeTools = toZod({
    tools: postgresToolkit.items,
    client: arcade,
    userId: "user@example.com"
});

// Create LangGraph tools
const tools = arcadeTools.map(({ name, description, execute, parameters }) =>
    tool(execute, {
        name,
        description,
        schema: parameters,
    })
);

// Setup model and agent
const model = new ChatOpenAI({
    model: "gpt-4o",
    apiKey: process.env.OPENAI_API_KEY
});
const boundModel = model.bindTools(tools);
const memory = new MemorySaver();

// Create agent graph
const graph = createReactAgent({
    llm: boundModel,
    tools,
    checkpointer: memory
});

// Execute database query
const config = {
    configurable: {
        thread_id: "postgres_session_1",
        user_id: "user@example.com"
    }
};

const userQuery = {
    messages: [
        { role: "user", content: "Analyze the monthly revenue trends from our sales table" }
    ]
};

// Stream results
const stream = await graph.stream(userQuery, config);
for await (const chunk of stream) {
    console.log(chunk.messages[chunk.messages.length - 1]);
}

Advanced Database Operations

Advanced Query Workflow Design

Create sophisticated database analysis workflows by combining multiple Postgres tools:

from langgraph.graph import END, START, StateGraph
from typing import TypedDict, List

class DatabaseAnalysisState(TypedDict):
    messages: List
    schema_info: dict
    query_results: List
    analysis: str

# Create workflow graph
workflow = StateGraph(DatabaseAnalysisState)

async def get_schema(state):
    """Retrieve database schema information"""
    tool_result = await tool_manager.execute_tool(
        "Postgres.GetSchema",
        {},
        user_id=state["user_id"]
    )
    return {"schema_info": tool_result}

async def analyze_data(state):
    """Run analysis queries based on schema"""
    # Build dynamic queries based on schema
    tables = state["schema_info"]["tables"]
    queries = []

    for table in tables[:5]:  # Analyze top 5 tables
        query_result = await tool_manager.execute_tool(
            "Postgres.QueryDatabase",
            {
                "query": f"SELECT COUNT(*) as row_count FROM {table['name']}"
            },
            user_id=state["user_id"]
        )
        queries.append(query_result)

    return {"query_results": queries}

async def generate_report(state):
    """Generate analysis report"""
    # Use LLM to analyze results
    analysis_prompt = f"""
    Based on the database schema: {state['schema_info']}
    And query results: {state['query_results']}

    Provide a comprehensive database analysis report.
    """

    response = await model.ainvoke(analysis_prompt)
    return {"analysis": response.content}

# Add nodes to workflow
workflow.add_node("get_schema", get_schema)
workflow.add_node("analyze_data", analyze_data)
workflow.add_node("generate_report", generate_report)

# Define flow
workflow.add_edge(START, "get_schema")
workflow.add_edge("get_schema", "analyze_data")
workflow.add_edge("analyze_data", "generate_report")
workflow.add_edge("generate_report", END)

# Compile and run
app = workflow.compile()

Implementing Safe Query Patterns

Ensure database operations remain secure and efficient:

class SafePostgresAgent:
    def __init__(self, tool_manager, max_rows=1000):
        self.tool_manager = tool_manager
        self.max_rows = max_rows

    async def safe_query(self, query: str, user_id: str):
        """Execute query with safety checks"""

        # Add row limit to prevent excessive data retrieval
        if "LIMIT" not in query.upper():
            query = f"{query} LIMIT {self.max_rows}"

        # Check for destructive operations
        forbidden_keywords = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
        if any(keyword in query.upper() for keyword in forbidden_keywords):
            return {"error": "Destructive operations not permitted"}

        # Execute with timeout
        try:
            result = await self.tool_manager.execute_tool(
                "Postgres.QueryDatabase",
                {"query": query, "timeout": 30},
                user_id=user_id
            )
            return result
        except Exception as e:
            return {"error": f"Query failed: {str(e)}"}

# Usage in LangGraph
safe_agent = SafePostgresAgent(tool_manager)

@tool
def safe_database_query(query: str, user_id: str) -> dict:
    """Safely query the Postgres database"""
    return safe_agent.safe_query(query, user_id)

Authentication and Multi-User Support

99% of MCP servers today are built for single-user use, even hosted ones. Arcade solves this by providing enterprise-grade multi-user authentication:

Per-User Database Access

class MultiUserDatabaseManager:
    def __init__(self, arcade_client):
        self.arcade = arcade_client
        self.user_connections = {}

    async def get_user_connection(self, user_id: str):
        """Get or create user-specific database connection"""

        if user_id not in self.user_connections:
            # Each user gets isolated database access
            auth_response = await self.arcade.auth.start(
                user_id=user_id,
                provider="postgres",
                scopes=["read", "write"]
            )

            if auth_response.status != "completed":
                return {
                    "authorization_required": True,
                    "url": auth_response.url
                }

            self.user_connections[user_id] = {
                "authenticated": True,
                "permissions": auth_response.scopes
            }

        return self.user_connections[user_id]

    async def execute_user_query(self, user_id: str, query: str):
        """Execute query with user-specific permissions"""

        connection = await self.get_user_connection(user_id)

        if connection.get("authorization_required"):
            return connection

        # Check user permissions
        if "write" not in connection["permissions"] and \
           any(op in query.upper() for op in ["INSERT", "UPDATE", "DELETE"]):
            return {"error": "User lacks write permissions"}

        # Execute with user context
        return await self.arcade.tools.execute(
            tool_name="Postgres.QueryDatabase",
            input={"query": query},
            user_id=user_id
        )

Production Deployment Strategies

Using Arcade Deploy

The fastest way to get started is with the arcade new command, which creates a complete MCP server project:

# Create new MCP server project
arcade new postgres-agent

# Navigate to project
cd postgres-agent

# Configure database connection
cat > .env << EOF
POSTGRES_HOST=your-host
POSTGRES_PORT=5432
POSTGRES_DATABASE=your-database
POSTGRES_USER=your-user
POSTGRES_PASSWORD=your-password
EOF

# Deploy to Arcade Cloud
arcade deploy

Your deployment configuration (worker.toml):

[worker]
name = "postgres-agent"
version = "1.0.0"
description = "Postgres database agent with LangGraph"

[worker.env]
required = [
    "POSTGRES_HOST",
    "POSTGRES_PORT",
    "POSTGRES_DATABASE",
    "POSTGRES_USER",
    "POSTGRES_PASSWORD"
]

[worker.tools]
enabled = ["Postgres.QueryDatabase", "Postgres.GetSchema", "Postgres.ListTables"]
max_concurrent = 10
timeout = 30

[worker.security]
ssl_required = true
connection_pooling = true
max_connections = 20

Kubernetes Deployment

For enterprise deployments:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: arcade-postgres-mcp
spec:
  replicas: 3
  selector:
    matchLabels:
      app: postgres-mcp-server
  template:
    metadata:
      labels:
        app: postgres-mcp-server
    spec:
      containers:
        - name: arcade-engine
          image: ghcr.io/arcadeai/engine:latest
          env:
            - name: ARCADE_API_KEY
              valueFrom:
                secretKeyRef:
                  name: arcade-secrets
                  key: api-key
            - name: POSTGRES_CONNECTION_STRING
              valueFrom:
                secretKeyRef:
                  name: postgres-secrets
                  key: connection-string
          ports:
            - containerPort: 8080
          resources:
            requests:
              memory: "512Mi"
              cpu: "500m"
            limits:
              memory: "1Gi"
              cpu: "1000m"
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 30
            periodSeconds: 10

Performance Optimization

Connection Pooling

Optimize database connections for high-concurrency scenarios:

class OptimizedPostgresManager:
    def __init__(self, arcade_client, pool_size=20):
        self.arcade = arcade_client
        self.pool_size = pool_size
        self.connection_cache = {}

    async def configure_pool(self):
        """Configure connection pooling"""
        await self.arcade.tools.configure(
            toolkit="Postgres",
            settings={
                "pool_size": self.pool_size,
                "max_overflow": 10,
                "pool_timeout": 30,
                "pool_recycle": 3600,
                "echo_pool": False
            }
        )

    async def cached_query(self, query_hash: str, query: str, user_id: str):
        """Execute query with result caching"""

        cache_key = f"{user_id}:{query_hash}"

        if cache_key in self.connection_cache:
            # Return cached result if fresh
            cached = self.connection_cache[cache_key]
            if cached["timestamp"] > time.time() - 300:  # 5 min cache
                return cached["result"]

        # Execute query
        result = await self.arcade.tools.execute(
            tool_name="Postgres.QueryDatabase",
            input={"query": query},
            user_id=user_id
        )

        # Cache result
        self.connection_cache[cache_key] = {
            "result": result,
            "timestamp": time.time()
        }

        return result

Monitoring and Observability

Track database operations and agent performance:

class DatabaseMonitor:
    def __init__(self, arcade_client):
        self.arcade = arcade_client
        self.metrics = {
            "query_count": 0,
            "error_count": 0,
            "avg_response_time": 0
        }

    async def monitored_query(self, query: str, user_id: str):
        """Execute query with monitoring"""

        start_time = time.time()

        try:
            result = await self.arcade.tools.execute(
                tool_name="Postgres.QueryDatabase",
                input={
                    "query": query,
                    "trace_id": str(uuid.uuid4())
                },
                user_id=user_id
            )

            # Update metrics
            self.metrics["query_count"] += 1
            response_time = time.time() - start_time
            self.update_avg_response_time(response_time)

            # Log to monitoring system
            await self.log_metric({
                "event": "database_query",
                "user_id": user_id,
                "response_time": response_time,
                "rows_returned": len(result.get("rows", []))
            })

            return result

        except Exception as e:
            self.metrics["error_count"] += 1
            await self.log_error({
                "event": "database_error",
                "user_id": user_id,
                "error": str(e),
                "query": query[:100]  # Log first 100 chars
            })
            raise

Best Practices and Troubleshooting

Query Optimization Tips

Use parameterized queries: Prevent SQL injection and improve performance
Implement query limits: Always include LIMIT clauses for safety
Cache frequently accessed data: Reduce database load with intelligent caching
Monitor query performance: Track slow queries and optimize indexes

Common Issues and Solutions

Connection Timeout Issues

# Increase timeout for long-running queries
result = await arcade.tools.execute(
    tool_name="Postgres.QueryDatabase",
    input={
        "query": complex_analytical_query,
        "timeout": 120  # 2 minutes for complex queries
    },
    user_id=user_id
)

Authentication Failures

# Handle auth gracefully
try:
    result = await execute_query(query, user_id)
except AuthorizationError as e:
    # Prompt user to re-authenticate
    auth_url = await arcade.auth.get_authorization_url(
        user_id=user_id,
        provider="postgres"
    )
    return {"auth_required": True, "url": auth_url}

Rate Limiting

# Implement exponential backoff
async def query_with_retry(query, user_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await execute_query(query, user_id)
        except RateLimitError:
            wait_time = (2 ** attempt) + random.random()
            await asyncio.sleep(wait_time)
    raise Exception("Max retries exceeded")

Next Steps

With Arcade's MCP support for Postgres integrated into your LangGraph application, you can:

Extend to multiple databases: Add MySQL, MongoDB, or ClickHouse support through Arcade's MCP server registry
Combine with other tools: Integrate Slack, Gmail, or GitHub alongside database operations
Build production agents: Deploy secure, multi-user database agents using Arcade Deploy
Create custom MCP servers: Build specialized database tools using the Arcade SDK

Conclusion

Querying Postgres from LangGraph through Arcade's MCP implementation eliminates the traditional barriers to database-enabled AI agents. By handling authentication, connection management, and security boundaries, Arcade lets you focus on building intelligent database workflows rather than infrastructure.

The combination of LangGraph's orchestration capabilities with Arcade's secure tool-calling platform enables production-ready agents that can safely interact with databases on behalf of multiple users. Whether you're building analytics agents, data pipeline automation, or intelligent reporting systems, this integration provides the foundation for sophisticated database operations.

Start building your database-enabled agents today with Arcade.dev and transform how your AI applications interact with data.

How to Query Postgres from LangGraph via Arcade (MCP)