	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server Caching with Redis

MCP Server Caching with Redis

Author: Venkata Sudhakar

MCP tool calls that fetch external data often return the same result for repeated identical inputs. Without caching, each call hits the downstream API, adding latency and cost. By adding a Redis cache layer inside the MCP server, you can serve repeated tool calls instantly from cache and only call the external API when the cache misses or expires.

In this tutorial, you will add Redis caching to an MCP server tool. The cache key is derived from the tool name and arguments. Results are stored with a configurable TTL. On a cache hit the server skips the external call entirely and returns the cached value.

The MCP server below uses the redis-py library to connect to a local or Cloud Memorystore Redis instance. The cache key is a hash of the tool arguments so identical requests always resolve to the same key.

# cached_server.py
import asyncio
import hashlib
import json
import time
import redis
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("cached-mcp-server")
cache = redis.Redis(host="localhost", port=6379, decode_responses=True)
DEFAULT_TTL = 300

def cache_key(tool_name, arguments):
    payload = json.dumps({"tool": tool_name, "args": arguments}, sort_keys=True)
    return "mcp:" + hashlib.sha256(payload.encode()).hexdigest()[:16]

def get_cached(key):
    return cache.get(key)

def set_cached(key, value, ttl=DEFAULT_TTL):
    cache.setex(key, ttl, value)

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_exchange_rate",
            description="Get the exchange rate between two currencies. Results cached for 5 minutes.",
            inputSchema={
                "type": "object",
                "properties": {
                    "from_currency": {"type": "string"},
                    "to_currency": {"type": "string"}
                },
                "required": ["from_currency", "to_currency"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    key = cache_key(name, arguments)
    cached = get_cached(key)
    if cached:
        return [TextContent(type="text", text="[CACHE HIT] " + cached)]
    if name == "get_exchange_rate":
        result = f"1 {arguments["from_currency"]} = 83.5 {arguments["to_currency"]}"
        set_cached(key, result)
        return [TextContent(type="text", text="[CACHE MISS] " + result)]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

The test below calls the same tool twice with identical arguments. The first call hits the external logic and populates the cache. The second call returns instantly from Redis without touching the external API.

Redis caching is most effective for tools that call external APIs with stable short-term results - exchange rates, weather data, product prices, or search index queries. Set the TTL based on how quickly the underlying data changes. For Cloud deployments, replace localhost Redis with a Cloud Memorystore instance and pass the connection details via environment variables.

Send your comments, suggestions or queries regarding this site to [email protected].