MCP Architecture: How the Model Context Protocol Works

Understanding MCP's technical architecture and working principles can help you better use and optimize MCP services. This article will deeply analyze MCP's core technologies.

---

1. Overall Architecture

1.1 Three-Layer Architecture

Application Layer:

• •User interaction interface
• •AI model reasoning
• •Tool call decision-making

Protocol Layer:

• •MCP Client: Built into AI applications
• •MCP Protocol: Based on JSON-RPC 2.0
• •MCP Server: Provides tools and resources

Service Layer:

• •Actual business logic
• •Data storage and queries
• •Permission control and security

1.2 Data Flow

---

2. Communication Protocol Details

2.1 JSON-RPC 2.0

MCP is based on JSON-RPC 2.0 protocol, a lightweight remote procedure call protocol.

Request Format:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query",
    "arguments": {
      "question": "Query order total",
      "role_id": null,
      "role_variables": null
    }
  }
}

Response Format:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "{\"status\": \"success\", \"data\": \"Order total is ¥123,456\"}"
      }
    ]
  }
}

2.2 Stdio Mode

Working Principle:

Characteristics:

• •Communicates through standard input/output
• •Each AI application starts an independent MCP Server process
• •Process lifecycle is bound to AI application

Startup Process:

# AI application starts MCP Server
$ uvx asktable-mcp-server@latest

# Environment variables
API_KEY=xxx
DATASOURCE_ID=yyy

# Communication
stdin  ← JSON-RPC request
stdout → JSON-RPC response
stderr → Log output

Advantages:

• •✅ Simple and direct, easy to debug
• •✅ Process isolation, good security
• •✅ No network configuration needed

Disadvantages:

• •❌ Each client has independent process, high resource usage
• •❌ Cannot communicate across machines
• •❌ Difficult to implement load balancing

2.3 SSE Mode

Working Principle:

Characteristics:

• •Based on HTTP and Server-Sent Events
• •Supports remote communication
• •Supports multi-client sharing

Connection Process:

SSE Event Format:

event: message
data: {"jsonrpc":"2.0","id":1,"result":{...}}

event: endpoint
data: http://localhost:8095/messages/

Advantages:

• •✅ Supports remote communication
• •✅ Multi-client sharing, high resource utilization
• •✅ Supports load balancing and high availability
• •✅ No local installation needed

Disadvantages:

• •❌ Network connection required
• •❌ Relatively complex configuration
• •❌ Need to handle network exceptions

---

3. Security Mechanisms

3.1 Authentication and Authorization

API Key Authentication:

Authentication Flow:

1. •Client carries API Key in request
2. •MCP Server forwards to AskTable API
3. •AskTable API verifies validity of API Key
4. •Execute query after verification passes

Credential Passing in SSE Mode:

In SSE mode, API Key is passed in two ways:

Method 1: URL Parameter (Recommended):

GET /sse/?api_key=xxx&datasource_id=yyy

Method 2: Session Storage:

# Store credentials on connection
pending_credentials[f"{client_ip}_{timestamp}"] = {
    "api_key": api_key,
    "datasource_id": datasource_id
}

# Associate with session on first request
session_credentials[session_id] = pending_credentials[key]

# Read from session on subsequent requests
credentials = session_credentials[session_id]

3.2 Permission Control

Row-Level Permissions:

# User query
query(
    question="Query orders",
    role_id="role_sales",
    role_variables={"region": "East China"}
)

# Actually executed SQL
SELECT * FROM orders
WHERE region = 'East China'  # Permission condition automatically injected

Permission Injection Flow:

Permission Rule Examples:

# Role definition
role = {
    "id": "role_sales",
    "rules": [
        "orders.region = {{region}}",
        "orders.status IN ('approved', 'completed')"
    ]
}

# Variable substitution
variables = {"region": "East China"}

# Final SQL
SELECT * FROM orders
WHERE region = 'East China'
  AND status IN ('approved', 'completed')

3.3 Data Security

Read-Only Access:

• •MCP Server only executes SELECT queries
• •Does not support INSERT, UPDATE, DELETE
• •Prevents accidental data modification

SQL Injection Prevention:

• •Uses parameterized queries
• •Does not directly concatenate user input
• •AST parsing and validation

Sensitive Data Masking:

• •Supports field-level masking
• •Automatically hides sensitive information (phone numbers, ID numbers, etc.)

---

4. Performance Optimization

4.1 Query Optimization

Smart Schema Linking:

Optimization Strategies:

• •Only query necessary fields (avoid SELECT *)
• •Automatically add LIMIT restrictions
• •Smart index selection
• •Avoid full table scans

Example:

# User question
"Query student names"

# Before optimization
SELECT * FROM students

# After optimization
SELECT name FROM students LIMIT 1000

4.2 Caching Mechanism

Multi-Layer Cache:

Cache Types:

• •Schema Cache: Table structure, field information
• •Vector Cache: Vector representation of tables and fields
• •Query Cache: Results of identical queries

Cache Strategy:

• •TTL: 1 hour
• •LRU eviction
• •Auto refresh

4.3 Connection Pool

Database Connection Pool:

# Connection pool configuration
pool = create_pool(
    min_size=5,      # Minimum connections
    max_size=20,     # Maximum connections
    timeout=30,      # Connection acquisition timeout
    recycle=3600     # Connection recycle time
)

Advantages:

• •Reduce connection establishment overhead
• •Reuse connections, improve performance
• •Control concurrency, protect database

4.4 Asynchronous Processing

Asynchronous Query:

async def query(question: str):
    # Asynchronously generate SQL
    sql = await generate_sql(question)

    # Asynchronously execute query
    result = await execute_query(sql)

    # Asynchronously format results
    formatted = await format_result(result)

    return formatted

Advantages:

• •Non-blocking I/O
• •Improved concurrency
• •Better resource utilization

---

5. AskTable MCP Server Implementation

5.1 Core Components

# server.py
from fastmcp import FastMCP

mcp = FastMCP(name="Asktable MCP Server")

@mcp.tool(name='Query data using AskTable')
async def query(
    question: str,
    role_id: str = None,
    role_variables: dict = None
) -> dict:
    # Call AskTable API
    result = await get_asktable_answer(
        api_key=os.getenv("API_KEY"),
        datasource_id=os.getenv("DATASOURCE_ID"),
        question=question,
        role_id=role_id,
        role_variables=role_variables
    )
    return result

5.2 API Calls

# at_apis.py
from asktable import Asktable

async def get_asktable_answer(
    api_key, datasource_id, question,
    base_url=None, role_id=None, role_variables=None
) -> dict:
    # Create client
    client = Asktable(api_key=api_key, base_url=base_url)

    # Call API
    response = client.answers.create(
        datasource_id=datasource_id,
        question=question,
        role_id=role_id,
        role_variables=role_variables
    )

    # Return result
    return {
        "status": "success" if response.answer else "failure",
        "data": response.answer.text if response.answer else None
    }

5.3 SSE Server

# sse_server.py
from fastapi import FastAPI
from fastmcp.server import FastMCPServer

app = FastAPI()

# Create MCP Server
mcp_server = FastMCPServer(mcp)

# SSE endpoint
@app.get("/sse/")
async def sse_endpoint(
    api_key: str,
    datasource_id: str
):
    # Store credentials
    store_credentials(api_key, datasource_id)

    # Return SSE stream
    return mcp_server.sse_handler()

# Message endpoint
@app.post("/messages/")
async def messages_endpoint(
    session_id: str,
    request: dict
):
    # Get credentials
    credentials = get_credentials(session_id)

    # Process request
    return await mcp_server.handle_request(request, credentials)

---

6. Monitoring and Debugging

6.1 Logging

Log Levels:

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

Key Logs:

• •Request logs: Record each query request
• •Performance logs: Record query duration
• •Error logs: Record exceptions and errors

Log Example:

2026-03-08 10:30:15 - INFO - Received query: Query order total
2026-03-08 10:30:16 - INFO - Generated SQL: SELECT SUM(amount) FROM orders
2026-03-08 10:30:17 - INFO - Query completed in 1.2s

6.2 Performance Monitoring

Key Metrics:

• •Request count (QPS)
• •Response time (P50, P95, P99)
• •Error rate
• •Concurrency

Monitoring Tools:

• •Prometheus + Grafana
• •ELK Stack
• •Custom monitoring

6.3 Debugging Tips

Stdio Mode Debugging:

# Manually run MCP Server
$ uvx asktable-mcp-server@latest

# Input JSON-RPC request
{"jsonrpc":"2.0","id":1,"method":"tools/list"}

# View response
{"jsonrpc":"2.0","id":1,"result":{...}}

SSE Mode Debugging:

# Test SSE connection
$ curl -N http://localhost:8095/sse/?api_key=xxx&datasource_id=yyy

# Test message endpoint
$ curl -X POST http://localhost:8095/messages/?session_id=abc \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'

---

7. Extension and Customization

7.1 Adding Custom Tools

@mcp.tool(name='Custom Tool')
async def custom_tool(param: str) -> dict:
    # Implement custom logic
    result = await do_something(param)
    return {"status": "success", "data": result}

7.2 Adding Resources

@mcp.resource("config://settings")
async def get_settings() -> str:
    # Return configuration information
    return json.dumps({"key": "value"})

7.3 Adding Prompt Templates

@mcp.prompt(name="Data Analysis Template")
async def analysis_prompt(topic: str) -> str:
    return f"Please analyze {topic}'s data, including trends, anomalies, and suggestions"

---

8. Summary

MCP's technical architecture reflects the following design principles:

Simplicity:

• •Based on standard protocol (JSON-RPC 2.0)
• •Supports multiple communication methods (Stdio, SSE)
• •Easy to integrate and extend

Security:

• •API Key authentication
• •Row-level permission control
• •Read-only access

Performance:

• •Smart query optimization
• •Multi-layer cache
• •Asynchronous processing

Scalability:

• •Plugin architecture
• •Supports custom tools
• •Rich ecosystem

Next Steps:

• •Practice more scenarios
• •Explore advanced features
• •Participate in community contributions

---

Related Reading:

• •MCP Introduction Guide
• •MCP Quick Start Guide
• •MCP Deployment Guide
• •MCP User Guide

Technical Exchange:

• •Website: asktable.com