Skip to main content

MCP Integration for Agents

Model Context Protocol (MCP) is how agents access tools, databases, and external data. This guide shows you how to use and build MCP servers.

Why Agents Need MCP

Without MCP:

Agent: "I need to check our database"
You: [Manually query database]
You: [Copy results to agent]
Agent: [Analyzes data]

With MCP:

Agent: "Check database for X"
→ Automatically queries via MCP
→ Gets results
→ Analyzes and reports

The difference: Autonomous tool access vs manual tool use.

MCP Architecture

MCP Server exposes 3 things:

  1. Tools - Functions agent can call
  2. Resources - Data agent can read
  3. Prompts - Pre-packaged instructions

Using Existing MCP Servers

Setup in Claude Desktop

1. Connect the server:

GitHub's official MCP server now runs as a local Docker container authenticated with a personal access token — the older npm-installed community package has been archived in favor of GitHub's own server. GitHub also runs a hosted, remote MCP endpoint (api.githubcopilot.com/mcp/), but as of 2026 it's restricted to GitHub Copilot integrations (VS Code, JetBrains, and similar) and rejects connections from third-party MCP clients like Claude Desktop.

2. Configure Claude:

Edit ~/Library/Application Support/Claude/claude_desktop_config.json:

{
"mcpServers": {
"github": {
"command": "docker",
"args": ["run", "-i", "--rm", "-e", "GITHUB_PERSONAL_ACCESS_TOKEN", "ghcr.io/github/github-mcp-server"],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "your-github-pat"
}
}
}
}

3. Restart Claude Desktop

4. Claude now has GitHub access:

You: "What open issues do we have in the main repo?"
Claude: [Uses MCP to query GitHub]
[Returns current issues]

Legacy Server Configs (No Longer Installable As-Is)

info

Anthropic's original reference servers for GitHub, Slack, Postgres, and Google Drive have been retired — check github.com/modelcontextprotocol/servers for the current actively-maintained list, and prefer official vendor servers (e.g., GitHub's own github-mcp-server) where available. The filesystem server remains one of the few original examples still maintained directly.

The Slack, PostgreSQL, and Google Drive configs below show the classic MCP config pattern — server name, command, args, env — but the specific @modelcontextprotocol/server-slack, server-postgres, and server-gdrive packages they reference are archived and no longer published as usable, maintained packages. Use them to understand the shape of an MCP server config, then swap in the current vendor-maintained replacement package for each service (check github.com/modelcontextprotocol/servers for the up-to-date list) before using them in a real setup.

Filesystem Access:

{
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/directory"]
}
}

Slack Integration:

Use Slack's own first-party MCP server rather than the retired community package — see Slack's MCP documentation for the current connection URL and OAuth setup.

PostgreSQL Database:

There's no single "official" Postgres MCP server — the retired @modelcontextprotocol/server-postgres package should not be used since it's no longer maintained or security-reviewed. If your Postgres is hosted on a platform like Supabase or AWS Aurora, use that vendor's first-party server. Otherwise, check github.com/modelcontextprotocol/servers (or the MCP registry) for a currently-maintained community option before wiring one in.

Google Drive:

Google now maintains its own official Workspace MCP server, which covers Drive, Gmail, and Calendar together — check Google's developer documentation for the current setup, rather than the retired @modelcontextprotocol/server-gdrive package.

Using Multiple Servers

Claude can use all configured servers simultaneously:

{
"mcpServers": {
"github": { /* config */ },
"postgres": { /* config */ },
"slack": { /* config */ },
"filesystem": { /* config */ }
}
}

Example task using multiple servers:

You: "Check GitHub issues, query the database for related bugs,
and post a summary to Slack"

Claude: [Uses GitHub MCP → Postgres MCP → Slack MCP]
[Completes entire workflow]

Building Your First MCP Server

Let's build a simple MCP server that exposes a tool.

Example: Weather Data Server (Python)

1. Install MCP SDK:

pip install mcp

2. Create server:

# weather_server.py
import asyncio
import httpx
from mcp.server.fastmcp import FastMCP

# Initialize server
mcp = FastMCP("weather-server")

# Define tool
@mcp.tool()
async def get_weather(
city: str,
units: str = "metric"
) -> str:
"""
Get current weather for a city.

Args:
city: City name (e.g., "San Francisco")
units: Temperature units ("metric" or "imperial")
"""
# Call weather API
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.openweathermap.org/data/2.5/weather",
params={
"q": city,
"units": units,
"appid": "your_api_key"
}
)
data = response.json()

# Format response
temp = data["main"]["temp"]
desc = data["weather"][0]["description"]
unit = "°C" if units == "metric" else "°F"

return f"{city}: {temp}{unit}, {desc}"

# Run server
if __name__ == "__main__":
mcp.run()

3. Configure in Claude:

{
"weather": {
"command": "python",
"args": ["/path/to/weather_server.py"]
}
}

4. Use it:

You: "What's the weather in Tokyo?"
Claude: [Calls get_weather via MCP]
"Tokyo: 18°C, partly cloudy"

Example: Custom Database Server

# database_server.py
import asyncio
import sqlite3
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("custom-db-server")

@mcp.tool()
async def query_customers(
status: str = None,
limit: int = 10
) -> list:
"""
Query customer database.

Args:
status: Filter by status (active, inactive, trial)
limit: Max results to return
"""
conn = sqlite3.connect('customers.db')
cursor = conn.cursor()

if status:
cursor.execute(
"SELECT * FROM customers WHERE status = ? LIMIT ?",
(status, limit)
)
else:
cursor.execute(
"SELECT * FROM customers LIMIT ?",
(limit,)
)

results = cursor.fetchall()
conn.close()

return results

@mcp.tool()
async def update_customer_status(
customer_id: int,
new_status: str
) -> str:
"""Update customer status."""
conn = sqlite3.connect('customers.db')
cursor = conn.cursor()

cursor.execute(
"UPDATE customers SET status = ? WHERE id = ?",
(new_status, customer_id)
)

conn.commit()
conn.close()

return f"Updated customer {customer_id} to {new_status}"

if __name__ == "__main__":
mcp.run()

Agent can now:

"Find all trial customers who signed up this month"
→ Agent queries database via MCP

"Upgrade customer 1234 to active status"
→ Agent updates via MCP

MCP Resources (Read-Only Data)

Resources are data the agent can read:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("docs-server")

@mcp.resource("docs://api-reference")
async def get_api_docs() -> str:
"""Provide API documentation"""
with open("api_docs.md", "r") as f:
return f.read()

@mcp.resource("docs://changelog")
async def get_changelog() -> str:
"""Provide version changelog"""
with open("CHANGELOG.md", "r") as f:
return f.read()

Agent usage:

You: "How do I use the authentication API?"
Claude: [Reads docs://api-reference via MCP]
[Provides answer from docs]

MCP Prompts (Pre-Packaged Instructions)

Prompts are reusable agent instructions:

@mcp.prompt()
async def analyze_user_feedback(
feedback_source: str
) -> str:
"""Analyze customer feedback for themes"""
return f"""
Analyze feedback from {feedback_source}:

1. Read all recent feedback entries
2. Identify common themes and pain points
3. Categorize by: Product, Service, Technical, Other
4. Calculate sentiment scores
5. Prioritize top 3 issues
6. Suggest improvements

Provide executive summary with action items.
"""

Agent usage:

You: "Run feedback analysis on support tickets"
Claude: [Loads analyze_user_feedback prompt]
[Executes all steps]
[Delivers report]

Advanced MCP Patterns

Pattern 1: Authentication

Secure MCP servers with auth:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("secure-server")

def verify_token(token: str) -> bool:
"""Verify API token"""
# Check token validity
return token == "expected_token"

@mcp.tool()
async def sensitive_operation(
token: str,
operation: str
) -> str:
"""Requires authentication"""
if not verify_token(token):
raise PermissionError("Invalid token")

# Proceed with operation
return f"Executed: {operation}"

Pattern 2: Rate Limiting

Prevent abuse:

import time
from collections import defaultdict

class RateLimiter:
def __init__(self, max_calls=10, period=60):
self.max_calls = max_calls
self.period = period
self.calls = defaultdict(list)

def check(self, identifier):
now = time.time()
cutoff = now - self.period

# Remove old calls
self.calls[identifier] = [
t for t in self.calls[identifier]
if t > cutoff
]

# Check limit
if len(self.calls[identifier]) >= self.max_calls:
raise Exception("Rate limit exceeded")

self.calls[identifier].append(now)

limiter = RateLimiter(max_calls=10, period=60)

@mcp.tool()
async def api_call(user_id: str, query: str):
limiter.check(user_id)
# Proceed with call

Pattern 3: Caching

Cache expensive operations:

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_api_call(query: str) -> str:
"""Cache results for repeated queries"""
# Expensive operation
result = expensive_api_call(query)
return result

@mcp.tool()
async def smart_search(query: str) -> str:
"""Search with caching"""
return cached_api_call(query)

Pattern 4: Batch Operations

Handle multiple items efficiently:

@mcp.tool()
async def batch_process(
items: list[str],
operation: str
) -> list:
"""Process multiple items in parallel"""
async with asyncio.TaskGroup() as group:
tasks = [
group.create_task(process_item(item, operation))
for item in items
]

return [t.result() for t in tasks]

Production MCP Best Practices

1. Error Handling

@mcp.tool()
async def robust_tool(param: str) -> str:
try:
result = risky_operation(param)
return result
except ConnectionError:
return "Service temporarily unavailable"
except ValueError as e:
return f"Invalid input: {e}"
except Exception as e:
# Log error
logger.error(f"Unexpected error: {e}")
return "An error occurred. Please try again."

2. Logging

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@mcp.tool()
async def logged_tool(param: str) -> str:
logger.info(f"Tool called with: {param}")

try:
result = operation(param)
logger.info(f"Success: {result}")
return result
except Exception as e:
logger.error(f"Failed: {e}")
raise

3. Input Validation

from pydantic import BaseModel, validator

class SearchParams(BaseModel):
query: str
limit: int = 10

@validator('query')
def query_not_empty(cls, v):
if not v.strip():
raise ValueError('Query cannot be empty')
return v

@validator('limit')
def limit_in_range(cls, v):
if not 1 <= v <= 100:
raise ValueError('Limit must be between 1 and 100')
return v

@mcp.tool()
async def validated_search(
query: str,
limit: int = 10
) -> list:
params = SearchParams(query=query, limit=limit)
return perform_search(params.query, params.limit)

4. Monitoring

import time
from prometheus_client import Counter, Histogram

# Metrics
tool_calls = Counter('mcp_tool_calls', 'Tool invocations')
tool_duration = Histogram('mcp_tool_duration', 'Tool execution time')

@mcp.tool()
async def monitored_tool(param: str) -> str:
tool_calls.inc()

start = time.time()
try:
result = operation(param)
return result
finally:
duration = time.time() - start
tool_duration.observe(duration)

MCP Server Deployment

Development

# Run locally
python mcp_server.py

Production

Option 1: Docker

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY mcp_server.py .

CMD ["python", "mcp_server.py"]

Option 2: Systemd Service

[Unit]
Description=MCP Server
After=network.target

[Service]
Type=simple
User=mcp
WorkingDirectory=/opt/mcp
ExecStart=/usr/bin/python3 /opt/mcp/server.py
Restart=always

[Install]
WantedBy=multi-user.target

Option 3: Cloud Function Deploy as serverless function on AWS Lambda, Google Cloud Functions, etc.

Testing MCP Servers

Unit Tests

import pytest
from mcp_server import mcp

@pytest.mark.asyncio
async def test_weather_tool():
result = await mcp.call_tool("get_weather", {
"city": "London",
"units": "metric"
})
assert "London" in result
assert "°C" in result

Integration Tests

@pytest.mark.asyncio
async def test_full_workflow():
# Test agent using MCP server
agent = Agent(mcp_servers=["weather"])

response = await agent.execute(
"What's the weather in Paris and Tokyo?"
)

assert "Paris" in response
assert "Tokyo" in response

MCP Registry & Discovery

Find servers:

Publish your server:

  1. Open source on GitHub
  2. Add mcp-server topic
  3. Submit to registry
  4. Share in communities

Common MCP Use Cases

Development:

  • GitHub, GitLab integration
  • Database query tools
  • CI/CD pipeline access
  • Log analysis tools

Business:

  • CRM integration (Salesforce, HubSpot)
  • Project management (Linear, Jira)
  • Communication (Slack, Teams)
  • Analytics (Google Analytics, Mixpanel)

Data:

  • SQL databases
  • NoSQL stores
  • Data warehouses
  • File systems

External Services:

  • Weather APIs
  • Maps and geocoding
  • Payment processing
  • Email services

Next Steps

  1. Try existing servers: Start with actively-maintained official servers (check github.com/modelcontextprotocol/servers for the current list, since several original Anthropic reference servers have since been retired)
  2. Build simple server: Create a tool for your use case
  3. Learn frameworks: Explore agent frameworks that support MCP
  4. Production deploy: Agent Safety for secure deployment

Resources:

MCP is what makes agents truly autonomous. Learn it well and your agents can do anything!