Developing an SSE-based Claude MCP shopping ai assistant service

Claude MCP supports two communication transport methods: STDIO (Standard Input/Output) or SSE (Server-Sent Events), both using JSON-RPC 2.0 for message formatting. STDIO is used for local integration, while SSE is used for network-based communication.

For example, if you need to use an MCP service directly in the command line, the STDIO transport method is appropriate. For web-based applications, the SSE transport method is the better choice.

In this article, we'll build an intelligent shopping assistant using an SSE-type MCP service with these key capabilities:

• Real-time product information and inventory access with custom order support
• Personalized product recommendations based on customer preferences and inventory availability
• Seamless interaction with microservices through the MCP tool server
• Live inventory checking when responding to product inquiries
• Streamlined product purchasing using product IDs and quantities
• Instant inventory updates after transactions
• Natural language analysis of order data

We'll use Anthropic Claude 3.5 Sonnet as our AI assistant for this MCP service, though any model supporting tool calling would work.

Our implementation requires several components:

1. A product microservice exposing API endpoints for product information
2. An order microservice with APIs for order creation and inventory management
3. An MCP SSE server that exposes microservice data to the LLM as tools via the SSE protocol
4. An MCP client that connects to the server through SSE to interact with the LLM

The complete project code is available at https://github.com/cnych/mcp-sse-demo

Microservices

Let's start by developing our product and order microservices with their API interfaces.

First, we'll define the types for products, inventory, and orders.

// types/index.ts
export interface Product {
  id: number;
  name: string;
  price: number;
  description: string;
}

export interface Inventory {
  productId: number;
  quantity: number;
  product?: Product;
}

export interface Order {
  id: number;
  customerName: string;
  items: Array<{ productId: number; quantity: number }>;
  totalAmount: number;
  orderDate: string;
}

Then we can use Express to expose the product and order microservices and provide API interfaces. Since it's a simulation, we'll use simpler in-memory data here, and expose the data through the following functions. (In a production environment, you still need to use a microservice plus a database to implement it.)

// services/product-service.ts
import { Product, Inventory, Order } from "../types/index.js";

// Simulate data storage
let products: Product[] = [
  {
    id: 1,
    name: "智能手表Galaxy",
    price: 1299,
    description: "健康监测，运动追踪，支持多种应用",
  },
  {
    id: 2,
    name: "无线蓝牙耳机Pro",
    price: 899,
    description: "主动降噪，30小时续航，IPX7防水",
  },
  {
    id: 3,
    name: "便携式移动电源",
    price: 299,
    description: "20000mAh大容量，支持快充，轻薄设计",
  },
  {
    id: 4,
    name: "华为MateBook X Pro",
    price: 1599,
    description: "14.2英寸全面屏，3:2比例，100% sRGB色域",
  },
];

// Simulate inventory data
let inventory: Inventory[] = [
  { productId: 1, quantity: 100 },
  { productId: 2, quantity: 50 },
  { productId: 3, quantity: 200 },
  { productId: 4, quantity: 150 },
];

let orders: Order[] = [];

export async function getProducts(): Promise<Product[]> {
  return products;
}

export async function getInventory(): Promise<Inventory[]> {
  return inventory.map((item) => {
    const product = products.find((p) => p.id === item.productId);
    return {
      ...item,
      product,
    };
  });
}

export async function getOrders(): Promise<Order[]> {
  return [...orders].sort(
    (a, b) => new Date(b.orderDate).getTime() - new Date(a.orderDate).getTime()
  );
}

export async function createPurchase(
  customerName: string,
  items: { productId: number; quantity: number }[]
): Promise<Order> {
  if (!customerName || !items || items.length === 0) {
    throw new Error("Invalid request: missing customer name or items");
  }

  let totalAmount = 0;

  // Verify inventory and calculate total price
  for (const item of items) {
    const inventoryItem = inventory.find((i) => i.productId === item.productId);
    const product = products.find((p) => p.id === item.productId);

    if (!inventoryItem || !product) {
      throw new Error(`Product ID ${item.productId} does not exist`);
    }

    if (inventoryItem.quantity < item.quantity) {
      throw new Error(
        `Product ${product.name} has insufficient inventory. Available: ${inventoryItem.quantity}`
      );
    }

    totalAmount += product.price * item.quantity;
  }

  // Create order
  const order: Order = {
    id: orders.length + 1,
    customerName,
    items,
    totalAmount,
    orderDate: new Date().toISOString(),
  };

  // Update inventory
  items.forEach((item) => {
    const inventoryItem = inventory.find(
      (i) => i.productId === item.productId
    )!;
    inventoryItem.quantity -= item.quantity;
  });

  orders.push(order);
  return order;
}

Then we can expose these API interfaces through MCP tools, as shown below:

// mcp-server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import {
  getProducts,
  getInventory,
  getOrders,
  createPurchase,
} from "./services/product-service.js";

export const server = new McpServer({
  name: "mcp-sse-demo",
  version: "1.0.0",
  description:
    "MCP tools for product query, inventory management, and order processing",
});

// Get product list tool
server.tool("getProducts", "Get all product information", {}, async () => {
  console.log("Getting product list");
  const products = await getProducts();
  return {
    content: [
      {
        type: "text",
        text: JSON.stringify(products),
      },
    ],
  };
});

// Get inventory information tool
server.tool(
  "getInventory",
  "Get all product inventory information",
  {},
  async () => {
    console.log("Getting inventory information");
    const inventory = await getInventory();
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(inventory),
        },
      ],
    };
  }
);

// Get order list tool
server.tool("getOrders", "Get all order information", {}, async () => {
  console.log("Getting order list");
  const orders = await getOrders();
  return {
    content: [
      {
        type: "text",
        text: JSON.stringify(orders),
      },
    ],
  };
});

// Purchase product tool
server.tool(
  "purchase",
  "Purchase product",
  {
    items: z
      .array(
        z.object({
          productId: z.number().describe("Product ID"),
          quantity: z.number().describe("Purchase quantity"),
        })
      )
      .describe("List of products to purchase"),
    customerName: z.string().describe("Customer name"),
  },
  async ({ items, customerName }) => {
    console.log("Processing purchase request", { items, customerName });
    try {
      const order = await createPurchase(customerName, items);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(order),
          },
        ],
      };
    } catch (error: any) {
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify({ error: error.message }),
          },
        ],
      };
    }
  }
);

Here we define 4 tools:

• getProducts: Retrieve all product information
• getInventory: Get inventory information for all products
• getOrders: Retrieve all order information
• purchase: Purchase products

If it's a Stdio-type MCP service, we can use these tools directly in the command line, but now we need to use an SSE-type MCP service, so we also need an MCP SSE server to expose these tools.

MCP SSE Server

Next, we'll develop an MCP SSE server to expose the product and order microservice data as tools using the SSE protocol.

// mcp-sse-server.ts
import express, { Request, Response, NextFunction } from "express";
import cors from "cors";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import { server as mcpServer } from "./mcp-server.js"; // Rename to avoid naming conflict

const app = express();
app.use(
  cors({
    origin: process.env.ALLOWED_ORIGINS?.split(",") || "*",
    methods: ["GET", "POST"],
    allowedHeaders: ["Content-Type", "Authorization"],
  })
);

// Store active connections
const connections = new Map();

// Health check endpoint
app.get("/health", (req, res) => {
  res.status(200).json({
    status: "ok",
    version: "1.0.0",
    uptime: process.uptime(),
    timestamp: new Date().toISOString(),
    connections: connections.size,
  });
});

// SSE connection establishment endpoint
app.get("/sse", async (req, res) => {
  // Instantiate SSE transport object
  const transport = new SSEServerTransport("/messages", res);
  // Get sessionId
  const sessionId = transport.sessionId;
  console.log(
    `[${new Date().toISOString()}] New SSE connection established: ${sessionId}`
  );

  // Register connection
  connections.set(sessionId, transport);

  // Connection interruption handling
  req.on("close", () => {
    console.log(
      `[${new Date().toISOString()}] SSE connection closed: ${sessionId}`
    );
    connections.delete(sessionId);
  });

  // Connect the transport object to the MCP server
  await mcpServer.connect(transport);
  console.log(
    `[${new Date().toISOString()}] MCP server connection successful: ${sessionId}`
  );
});

// Endpoint for receiving client messages
app.post("/messages", async (req: Request, res: Response) => {
  try {
    console.log(
      `[${new Date().toISOString()}] Received client message:`,
      req.query
    );
    const sessionId = req.query.sessionId as string;

    // Find the corresponding SSE connection and process the message
    if (connections.size > 0) {
      const transport: SSEServerTransport = connections.get(
        sessionId
      ) as SSEServerTransport;
      // Use transport to process messages
      if (transport) {
        await transport.handlePostMessage(req, res);
      } else {
        throw new Error("No active SSE connection");
      }
    } else {
      throw new Error("No active SSE connection");
    }
  } catch (error: any) {
    console.error(
      `[${new Date().toISOString()}] Failed to process client message:`,
      error
    );
    res
      .status(500)
      .json({ error: "Failed to process message", message: error.message });
  }
});

// Graceful shutdown of all connections
async function closeAllConnections() {
  console.log(
    `[${new Date().toISOString()}] Closing all connections (${
      connections.size
    }个)`
  );
  for (const [id, transport] of connections.entries()) {
    try {
      // Send shutdown event
      transport.res.write(
        'event: server_shutdown\ndata: {"reason": "Server is shutting down"}\n\n'
      );
      transport.res.end();
      console.log(`[${new Date().toISOString()}] Connection closed: ${id}`);
    } catch (error) {
      console.error(
        `[${new Date().toISOString()}] Failed to close connection: ${id}`,
        error
      );
    }
  }
  connections.clear();
}

// Error handling
app.use((err: Error, req: Request, res: Response, next: NextFunction) => {
  console.error(`[${new Date().toISOString()}] Unhandled exception:`, err);
  res.status(500).json({ error: "Server internal error" });
});

// Graceful shutdown
process.on("SIGTERM", async () => {
  console.log(
    `[${new Date().toISOString()}] Received SIGTERM signal, preparing to close`
  );
  await closeAllConnections();
  server.close(() => {
    console.log(`[${new Date().toISOString()}] Server closed`);
    process.exit(0);
  });
});

process.on("SIGINT", async () => {
  console.log(
    `[${new Date().toISOString()}] Received SIGINT signal, preparing to close`
  );
  await closeAllConnections();
  process.exit(0);
});

// Start server
const port = process.env.PORT || 8083;
const server = app.listen(port, () => {
  console.log(
    `[${new Date().toISOString()}] Smart shopping MCP SSE server started, address: http://localhost:${port}`
  );
  console.log(`- SSE connection endpoint: http://localhost:${port}/sse`);
  console.log(
    `- Message processing endpoint: http://localhost:${port}/messages`
  );
  console.log(`- Health check endpoint: http://localhost:${port}/health`);
});

Here we use Express to expose an SSE connection endpoint /sse, for receiving client messages. We use SSEServerTransport to create an SSE transport object and specify the message processing endpoint as /messages.

const transport = new SSEServerTransport("/messages", res);

After the transport object is created, we can connect the transport object to the MCP server, as shown below:

// Connect the transport object to the MCP server
await mcpServer.connect(transport);

Then we can receive client messages through the SSE connection endpoint /sse, and use the message processing endpoint /messages to process client messages. When a client message is received, we need to use the transport object to handle the client message in the /messages endpoint:

// Use transport to process messages
await transport.handlePostMessage(req, res);

This is the same as the operation of listing tools and calling tools.

MCP Client

Next, we'll develop an MCP client to connect to the MCP SSE server and interact with the LLM. We can develop a command line client or a web client.

For the command line client, we've already introduced it, the only difference is that now we need to use the SSE protocol to connect to the MCP SSE server.

// Create MCP client
const mcpClient = new McpClient({
  name: "mcp-sse-demo",
  version: "1.0.0",
});

// Create SSE transport object
const transport = new SSEClientTransport(new URL(config.mcp.serverUrl));

// Connect to MCP server
await mcpClient.connect(transport);

After establishing the connection, the workflow mirrors our previous command line client implementation: we enumerate available tools, forward the user's query along with tool definitions to the LLM, process any tool invocations from the LLM response, and finally send the tool execution results back to the LLM to generate the comprehensive answer.

For our web-based implementation, the core logic remains identical to the command line version, with the key difference being that we now expose these processing steps through API endpoints that our web frontend can interact with.

The initialization process begins with setting up the MCP client, retrieving the available tools, transforming them into Anthropic's expected format (as their API requires a specific schema structure), and finally instantiating the Anthropic client that will handle our AI interactions.

// Initialize MCP client
async function initMcpClient() {
  if (mcpClient) return;

  try {
    console.log("Connecting to MCP server...");
    mcpClient = new McpClient({
      name: "mcp-client",
      version: "1.0.0",
    });

    const transport = new SSEClientTransport(new URL(config.mcp.serverUrl));

    await mcpClient.connect(transport);
    const { tools } = await mcpClient.listTools();
    // Convert tool format to the array form required by Anthropic
    anthropicTools = tools.map((tool: any) => {
      return {
        name: tool.name,
        description: tool.description,
        input_schema: tool.inputSchema,
      };
    });
    // Create Anthropic client
    aiClient = createAnthropicClient(config);

    console.log("MCP client and tools initialized");
  } catch (error) {
    console.error("Failed to initialize MCP client:", error);
    throw error;
  }
}

Then we can develop API interfaces according to our own needs, for example, we develop a chat interface here to receive the user's question, then call the MCP client's tools, send the tool call results and history messages back to the LLM for processing to get the final result, as shown below:

// API: Chat request
apiRouter.post("/chat", async (req, res) => {
  try {
    const { message, history = [] } = req.body;

    if (!message) {
      console.warn("请求中消息为空");
      return res.status(400).json({ error: "消息不能为空" });
    }

    // Build message history
    const messages = [...history, { role: "user", content: message }];

    // Call AI
    const response = await aiClient.messages.create({
      model: config.ai.defaultModel,
      messages,
      tools: anthropicTools,
      max_tokens: 1000,
    });

    // Process tool calls
    const hasToolUse = response.content.some(
      (item) => item.type === "tool_use"
    );

    if (hasToolUse) {
      // Process all tool calls
      const toolResults = [];

      for (const content of response.content) {
        if (content.type === "tool_use") {
          const name = content.name;
          const toolInput = content.input as
            | { [x: string]: unknown }
            | undefined;

          try {
            // Call MCP tool
            if (!mcpClient) {
              console.error("MCP client not initialized");
              throw new Error("MCP client not initialized");
            }
            console.log(`Calling MCP tool: ${name}`);
            const toolResult = await mcpClient.callTool({
              name,
              arguments: toolInput,
            });

            toolResults.push({
              name,
              result: toolResult,
            });
          } catch (error: any) {
            console.error(`Tool call failed: ${name}`, error);
            toolResults.push({
              name,
              error: error.message,
            });
          }
        }
      }

      // Send tool results back to AI to get the final response
      const finalResponse = await aiClient.messages.create({
        model: config.ai.defaultModel,
        messages: [
          ...messages,
          {
            role: "user",
            content: JSON.stringify(toolResults),
          },
        ],
        max_tokens: 1000,
      });

      const textResponse = finalResponse.content
        .filter((c) => c.type === "text")
        .map((c) => c.text)
        .join("\n");

      res.json({
        response: textResponse,
        toolCalls: toolResults,
      });
    } else {
      // Return AI response directly
      const textResponse = response.content
        .filter((c) => c.type === "text")
        .map((c) => c.text)
        .join("\n");

      res.json({
        response: textResponse,
        toolCalls: [],
      });
    }
  } catch (error: any) {
    console.error("Chat request processing failed:", error);
    res.status(500).json({ error: error.message });
  }
});

The core implementation is also relatively simple, and is basically the same as the command line client, except that now we implement these processing steps in some interfaces.

Usage

Here is an example of using the command line client:

We can also use it in Cursor, create a .cursor/mcp.json file, and add the following content:

{
  "mcpServers": {
    "products-sse": {
      "url": "http://localhost:8083/sse"
    }
  }
}

Then we can see this MCP service in the Cursor settings page, and then we can use this MCP service in Cursor.

Here is an example of using the web client we developed:

Summary

When implementing tool calling functionality with LLMs, the effectiveness largely depends on well-crafted tool descriptions:

• Clear Documentation: Each tool should have precise, unambiguous descriptions with strategic keywords that help the model determine appropriate usage contexts
• Functional Distinction: Tools should have clearly differentiated purposes to prevent the LLM from making incorrect selections between similar options
• Comprehensive Testing: Before deployment, validate tool calling accuracy across diverse user query patterns to ensure reliable performance

The MCP server architecture offers implementation flexibility through multiple technology options:

• Python SDK implementation
• TypeScript/JavaScript frameworks
• Various other programming language options

Technology selection should align with your team's expertise and complement your existing infrastructure.

Incorporating an AI assistant with MCP capabilities into a microservice ecosystem delivers several strategic benefits:

1. Dynamic Data Delivery: SSE protocol enables immediate data transmission, critical for time-sensitive information like inventory levels and order tracking
2. Independent Scaling: System components can be scaled according to their specific demand patterns, optimizing resource allocation
3. Fault Isolation: The distributed architecture prevents cascading failures, as individual service disruptions remain contained
4. Technical Diversity: Teams can develop and maintain different system components using their preferred technology stacks
5. Bandwidth Optimization: SSE's event-driven approach reduces network overhead compared to polling methods by transmitting only when changes occur
6. Responsive Interfaces: Immediate updates and rapid system responses significantly enhance the customer experience
7. Streamlined Frontend: Client implementations become more elegant without complex polling logic, simply subscribing to server events

For production deployments, several additional considerations become important:

• Implement rigorous error detection through comprehensive testing protocols
• Establish robust failover mechanisms to maintain service continuity
• Deploy performance monitoring systems that track tool calling accuracy and response times
• Evaluate caching strategies to minimize backend service load during peak usage

By following these architectural principles, you can create a high-performance, reliable MCP-powered shopping assistant that delivers personalized, real-time experiences to your customers.

Origin article link: https://www.claudemcp.com/docs/dev-sse-mcp