Claude MCP × Agent Workflows: Designing and Building Real-World Automation Systems

mkdir my-mcp-server
cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node ts-node

import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
 
const server = new McpServer({
  name: "data-analyzer",
  version: "1.0.0",
});
 
// Define an aggregation tool
server.tool(
  "aggregate_data",
  "Aggregate CSV data and return statistics",
  {
    data: z.array(z.record(z.string(), z.number())).describe("Array of data objects to aggregate"),
    column: z.string().describe("Column name to aggregate"),
    operation: z.enum(["sum", "average", "max", "min"]).describe("Aggregation operation"),
  },
  async ({ data, column, operation }) => {
    const values = data.map(row => row[column]).filter(v => v \!== undefined);
    
    let result: number;
    switch (operation) {
      case "sum": result = values.reduce((a, b) => a + b, 0); break;
      case "average": result = values.reduce((a, b) => a + b, 0) / values.length; break;
      case "max": result = Math.max(...values); break;
      case "min": result = Math.min(...values); break;
    }
    
    return {
      content: [{
        type: "text",
        text: JSON.stringify({ column, operation, result, count: values.length }),
      }],
    };
  }
);
 
// Define a resource with a template URI
server.resource(
  "report",
  new ResourceTemplate("report://{date}", { list: undefined }),
  async (uri, { date }) => ({
    contents: [{
      uri: uri.href,
      text: `Report data for ${date} (sample)`,
    }],
  })
);
 
// Start with STDIO transport
const transport = new StdioServerTransport();
await server.connect(transport);

{
  "mcpServers": {
    "data-analyzer": {
      "command": "node",
      "args": ["/path/to/my-mcp-server/dist/index.js"],
      "env": {
        "NODE_ENV": "production"
      }
    }
  }
}

from anthropic import Anthropic
 
client = Anthropic()
 
def daily_report_workflow(sources: list[str]) -> str:
    """Generate a daily digest"""
    
    collection_prompt = f"""
    Collect today's important news from the following sources:
    {', '.join(sources)}
    
    For each source:
    1. Use fetch_webpage to retrieve the page
    2. Extract the 5 most recent article titles and summaries
    3. Rate their importance from 1–5
    """
    
    collection_result = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=8192,
        messages=[{"role": "user", "content": collection_prompt}]
    )
    
    summary_prompt = f"""
    Analyze the collected content and create an executive summary for today.
    
    Collected content:
    {collection_result.content[0].text}
    
    Requirements:
    - Lead with the 3 most important topics
    - Keep each topic to 3 lines or fewer
    - Include industry impact and recommended actions
    - Use Markdown formatting optimized for Slack readability
    """
    
    summary = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": summary_prompt}]
    )
    
    return summary.content[0].text

def triage_inquiry(email_content: str) -> dict:
    """Triage a support inquiry"""
    
    triage_result = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""You are a customer support triage specialist.
        Classify the inquiry into one of the following categories and assess urgency:
        - billing: payment or billing issues
        - technical: technical problems
        - general: general questions
        - complaint: complaints or dissatisfaction
        
        Urgency: high (immediate human response) / medium (within 4 hours) / low (within 24 hours)
        
        Respond in JSON format.""",
        messages=[{"role": "user", "content": email_content}]
    )
    
    return json.loads(triage_result.content[0].text)

import time
import random
from typing import Callable, TypeVar
 
T = TypeVar('T')
 
def with_retry(
    func: Callable[[], T],
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> T:
    """Retry with exponential backoff and jitter"""
    
    for attempt in range(max_attempts):
        try:
            return func()
        except Exception as e:
            if attempt == max_attempts - 1:
                raise
            
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s...")
            time.sleep(delay)

import logging
import json
from datetime import datetime
 
class AgentLogger:
    def __init__(self, workflow_id: str):
        self.workflow_id = workflow_id
        self.logger = logging.getLogger(f"agent.{workflow_id}")
    
    def log_tool_call(self, tool_name: str, input_data: dict, output_data: dict, duration_ms: float):
        self.logger.info(json.dumps({
            "timestamp": datetime.utcnow().isoformat(),
            "workflow_id": self.workflow_id,
            "event": "tool_call",
            "tool": tool_name,
            "input_tokens": len(str(input_data)),
            "output_tokens": len(str(output_data)),
            "duration_ms": duration_ms,
        }))

response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=4096,
    system=[
        {
            "type": "text",
            "text": "(long system prompt...)",
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[{"role": "user", "content": user_message}]
)

def summarize_history(messages: list, threshold: int = 20) -> list:
    """Compress old messages into a summary"""
    if len(messages) <= threshold:
        return messages
    
    old_messages = messages[:-threshold]
    recent_messages = messages[-threshold:]
    
    summary_response = client.messages.create(
        model="claude-haiku-4-5",  # Haiku is sufficient for summarization
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Summarize the following conversation history concisely:\n\n{json.dumps(old_messages)}"
        }]
    )
    
    summary = summary_response.content[0].text
    return [{"role": "assistant", "content": f"[Conversation summary]: {summary}"}] + recent_messages

def sanitize_user_input(user_input: str) -> str:
    """Sanitize user input"""
    suspicious_patterns = [
        "ignore previous instructions",
        "system prompt",
        "forget everything",
    ]
    
    lower_input = user_input.lower()
    for pattern in suspicious_patterns:
        if pattern in lower_input:
            raise ValueError(f"Suspicious input detected: {pattern}")
    
    if len(user_input) > 10000:
        raise ValueError("Input too long")
    
    return user_input