Post

Model Context Protocol: The Future of AI-Native Web Applications

Posted
By Shivraj Badu 12 min read

Model Context Protocol: The Future of AI-Native Web Applications

Introduction

The Model Context Protocol (MCP) represents a fundamental shift in how AI models interact with applications. Instead of applications calling AI models, AI models call application resources. This inversion of control transforms how we build AI-native software.

Announced by Anthropic in late 2024 and rapidly adopted across the industry, MCP solves a problem every AI application faces: how do you give language models secure, standardized access to external context—files, databases, APIs, real-time data—without building custom integrations for every tool?

MCP is to AI applications what REST was to web services: a universal protocol that makes integration predictable, discoverable, and composable. Understanding MCP isn’t optional for AI engineers in 2026—it’s foundational.

The Problem MCP Solves

Before MCP:

Every AI application builder faced the same challenges:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# Custom integration hell
def answer_question(question: str):
    # Custom file access
    files = read_company_files()
    
    # Custom database query
    data = query_custom_database()
    
    # Custom API integration
    external_data = call_proprietary_api()
    
    # Custom tool calling logic
    if "search" in question:
        results = custom_search()
    
    # Combine everything manually
    context = f"{files}\n{data}\n{external_data}\n{results}"
    
    return llm.generate(question, context=context)

Problems:

  • Every tool requires custom integration
  • No standardization across providers
  • Security model unclear
  • No discoverability
  • Can’t compose tools easily

After MCP:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# Standardized MCP integration
from mcp import Client

client = Client()

# Discover available resources
resources = client.list_resources()

# Access any resource through standard protocol
context = client.read_resource("files://company/policies")

# Call any tool through standard protocol
result = client.call_tool("search_database", {"query": "revenue"})

# LLM automatically discovers and uses available context
answer = llm.generate(question, mcp_client=client)

MCP Architecture

Core Concepts

1. Servers (Context Providers)

MCP servers expose resources and tools to AI models:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

from mcp.server import Server
from mcp.types import Resource, Tool

class FileSystemServer(Server):
    """Expose file system access via MCP"""
    
    def list_resources(self) -> list[Resource]:
        """Advertise available resources"""
        return [
            Resource(
                uri="files://documents/",
                name="Company Documents",
                description="Internal documentation",
                mime_type="text/plain"
            )
        ]
    
    def read_resource(self, uri: str) -> str:
        """Read resource content"""
        if uri.startswith("files://"):
            path = uri.replace("files://", "")
            return self._read_file_safely(path)
        raise ValueError(f"Unknown resource: {uri}")
    
    def list_tools(self) -> list[Tool]:
        """Advertise available tools"""
        return [
            Tool(
                name="search_files",
                description="Search for files by content",
                input_schema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"}
                    },
                    "required": ["query"]
                }
            )
        ]
    
    def call_tool(self, name: str, arguments: dict) -> any:
        """Execute tool"""
        if name == "search_files":
            return self._search_files(arguments["query"])
        raise ValueError(f"Unknown tool: {name}")

2. Clients (AI Applications)

MCP clients connect to servers and make context available to LLMs:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from mcp import Client
from anthropic import Anthropic

class MCPClient:
    def __init__(self, server_url: str):
        self.mcp = Client(server_url)
        self.llm = Anthropic()
    
    async def query_with_context(self, question: str) -> str:
        """Query LLM with MCP context"""
        # Discover available resources
        resources = await self.mcp.list_resources()
        
        # Discover available tools
        tools = await self.mcp.list_tools()
        
        # LLM decides what context to fetch
        response = await self.llm.messages.create(
            model="claude-3-opus",
            messages=[{"role": "user", "content": question}],
            mcp_client=self.mcp,  # Pass MCP client
            tools=tools  # LLM can call these tools
        )
        
        return response.content

3. Protocol Flow

1
2
3
4
5
6
7
8
9

┌─────────┐           ┌──────────┐           ┌────────────┐
│   LLM   │◄─────────►│ MCP      │◄─────────►│ MCP Server │
│ (Claude)│   Query   │ Client   │  Protocol │ (Your App) │
└─────────┘           └──────────┘           └────────────┘
                           │
                           ├─ list_resources()
                           ├─ read_resource(uri)
                           ├─ list_tools()
                           └─ call_tool(name, args)

Building an MCP Server

Example: Database MCP Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

from mcp.server import Server
from mcp.types import Resource, Tool
import asyncpg
from typing import Dict, List

class DatabaseMCPServer(Server):
    """Expose PostgreSQL database via MCP"""
    
    def __init__(self, database_url: str):
        self.pool = None
        self.database_url = database_url
    
    async def startup(self):
        """Initialize connection pool"""
        self.pool = await asyncpg.create_pool(self.database_url)
    
    async def list_resources(self) -> List[Resource]:
        """Expose database tables as resources"""
        async with self.pool.acquire() as conn:
            tables = await conn.fetch("""
                SELECT table_name 
                FROM information_schema.tables 
                WHERE table_schema = 'public'
            """)
        
        return [
            Resource(
                uri=f"db://table/{table['table_name']}",
                name=table['table_name'],
                description=f"Database table: {table['table_name']}",
                mime_type="application/json"
            )
            for table in tables
        ]
    
    async def read_resource(self, uri: str) -> str:
        """Read table schema or sample data"""
        if uri.startswith("db://table/"):
            table_name = uri.split("/")[-1]
            
            async with self.pool.acquire() as conn:
                # Get schema
                schema = await conn.fetch("""
                    SELECT column_name, data_type 
                    FROM information_schema.columns 
                    WHERE table_name = $1
                """, table_name)
                
                # Get sample rows
                rows = await conn.fetch(f"SELECT * FROM {table_name} LIMIT 5")
            
            return {
                "schema": [dict(col) for col in schema],
                "sample_data": [dict(row) for row in rows]
            }
    
    async def list_tools(self) -> List[Tool]:
        """Expose database operations as tools"""
        return [
            Tool(
                name="query_database",
                description="Execute a read-only SQL query",
                input_schema={
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "SQL SELECT query"
                        }
                    },
                    "required": ["query"]
                }
            ),
            Tool(
                name="get_table_summary",
                description="Get statistical summary of a table",
                input_schema={
                    "type": "object",
                    "properties": {
                        "table_name": {"type": "string"}
                    },
                    "required": ["table_name"]
                }
            )
        ]
    
    async def call_tool(self, name: str, arguments: Dict) -> any:
        """Execute database tool"""
        if name == "query_database":
            return await self._execute_query(arguments["query"])
        elif name == "get_table_summary":
            return await self._table_summary(arguments["table_name"])
    
    async def _execute_query(self, query: str) -> List[Dict]:
        """Safely execute read-only query"""
        # Security: Only allow SELECT
        if not query.strip().upper().startswith("SELECT"):
            raise ValueError("Only SELECT queries allowed")
        
        async with self.pool.acquire() as conn:
            rows = await conn.fetch(query)
            return [dict(row) for row in rows]
    
    async def _table_summary(self, table_name: str) -> Dict:
        """Get table statistics"""
        async with self.pool.acquire() as conn:
            count = await conn.fetchval(f"SELECT COUNT(*) FROM {table_name}")
            
            # Get numeric column stats
            columns = await conn.fetch("""
                SELECT column_name 
                FROM information_schema.columns 
                WHERE table_name = $1 AND data_type IN ('integer', 'numeric', 'double precision')
            """, table_name)
            
            stats = {}
            for col in columns:
                col_name = col['column_name']
                stat = await conn.fetchrow(f"""
                    SELECT 
                        MIN({col_name}) as min,
                        MAX({col_name}) as max,
                        AVG({col_name}) as avg
                    FROM {table_name}
                """)
                stats[col_name] = dict(stat)
        
        return {
            "row_count": count,
            "column_statistics": stats
        }

# Run the server
if __name__ == "__main__":
    server = DatabaseMCPServer("postgresql://localhost/mydb")
    server.run(host="0.0.0.0", port=8000)

Using MCP in Applications

Rails Integration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# app/services/mcp_service.rb
class MCPService
  def initialize
    @client = MCP::Client.new(
      servers: {
        database: ENV['MCP_DATABASE_SERVER'],
        files: ENV['MCP_FILES_SERVER'],
        api: ENV['MCP_API_SERVER']
      }
    )
  end
  
  def query_with_context(question)
    # Fetch relevant context
    resources = @client.list_resources
    relevant = resources.select { |r| relevant_to?(r, question) }
    
    context = relevant.map { |r| @client.read_resource(r.uri) }.join("\n")
    
    # Query LLM with context
    response = anthropic_client.messages(
      model: "claude-3-opus",
      messages: [{ role: "user", content: question }],
      system: "Use the provided context to answer. Context:\n#{context}"
    )
    
    response.content
  end
  
  private
  
  def relevant_to?(resource, question)
    # Simple relevance check (could use embeddings)
    keywords = extract_keywords(question)
    keywords.any? { |kw| resource.name.downcase.include?(kw) }
  end
end

# app/controllers/api/questions_controller.rb
class Api::QuestionsController < ApplicationController
  def create
    mcp = MCPService.new
    answer = mcp.query_with_context(params[:question])
    
    render json: { answer: answer }
  end
end

Python FastAPI Integration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from mcp import Client
from anthropic import Anthropic

app = FastAPI()

class QuestionRequest(BaseModel):
    question: str
    user_id: str

class AnswerResponse(BaseModel):
    answer: str
    sources: list[str]

@app.on_event("startup")
async def startup():
    """Initialize MCP connections"""
    app.state.mcp = Client()
    
    # Register MCP servers
    await app.state.mcp.add_server("database", "http://localhost:8000")
    await app.state.mcp.add_server("docs", "http://localhost:8001")
    
    app.state.anthropic = Anthropic()

@app.post("/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    """Answer question using MCP context"""
    try:
        # Discover available tools
        tools = await app.state.mcp.list_tools()
        
        # Query with MCP context
        message = await app.state.anthropic.messages.create(
            model="claude-3-opus",
            messages=[{"role": "user", "content": request.question}],
            tools=tools,
            max_tokens=1024
        )
        
        # Extract answer and tool calls
        answer = message.content[0].text if message.content else ""
        tool_calls = [
            block.name 
            for block in message.content 
            if hasattr(block, 'name')
        ]
        
        return AnswerResponse(
            answer=answer,
            sources=tool_calls
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Real-World MCP Servers

1. GitHub MCP Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class GitHubMCPServer(Server):
    """Access GitHub repos via MCP"""
    
    async def list_resources(self):
        return [
            Resource(
                uri="github://repo/owner/name",
                name="Repository Code",
                description="Source code from GitHub"
            ),
            Resource(
                uri="github://issues/owner/name",
                name="Repository Issues",
                description="GitHub issues and PRs"
            )
        ]
    
    async def list_tools(self):
        return [
            Tool(name="search_code", ...),
            Tool(name="create_issue", ...),
            Tool(name="get_pr_diff", ...)
        ]

2. Slack MCP Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

class SlackMCPServer(Server):
    """Access Slack workspace via MCP"""
    
    async def list_resources(self):
        channels = await self.slack.conversations_list()
        
        return [
            Resource(
                uri=f"slack://channel/{channel['id']}",
                name=f"#{channel['name']}",
                description=f"Slack channel: {channel['purpose']}"
            )
            for channel in channels
        ]
    
    async def list_tools(self):
        return [
            Tool(name="search_messages", ...),
            Tool(name="post_message", ...),
            Tool(name="get_user_info", ...)
        ]

3. Google Drive MCP Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

class GoogleDriveMCPServer(Server):
    """Access Google Drive files via MCP"""
    
    async def read_resource(self, uri: str):
        file_id = uri.split("/")[-1]
        
        # Download file
        content = self.drive.files().get_media(fileId=file_id).execute()
        
        # Extract text
        if file_id.endswith('.pdf'):
            text = self._extract_pdf_text(content)
        elif file_id.endswith('.docx'):
            text = self._extract_docx_text(content)
        else:
            text = content.decode()
        
        return text

Security Considerations

Authentication and Authorization

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from mcp.server import Server
from mcp.auth import require_auth, check_permissions

class SecureMCPServer(Server):
    @require_auth
    async def read_resource(self, uri: str, auth_context: dict):
        """Only authenticated users can read resources"""
        user_id = auth_context['user_id']
        
        # Check permissions
        if not await self._user_can_access(user_id, uri):
            raise PermissionError("Access denied")
        
        return await self._read_resource_internal(uri)
    
    @require_auth
    @check_permissions("database:write")
    async def call_tool(self, name: str, arguments: dict, auth_context: dict):
        """Tool calls require specific permissions"""
        if name == "delete_record":
            # Extra caution for destructive operations
            await self._audit_log(auth_context['user_id'], name, arguments)
        
        return await self._call_tool_internal(name, arguments)

Rate Limiting

1
2
3
4
5
6
7
8
9
10

from mcp.middleware import RateLimiter

server = Server()
server.add_middleware(
    RateLimiter(
        max_requests=100,
        window_seconds=60,
        key_func=lambda req: req.auth_context['user_id']
    )
)

Input Validation

1
2
3
4
5
6
7
8
9
10
11
12
13
14

class ValidatedMCPServer(Server):
    async def call_tool(self, name: str, arguments: dict):
        # Validate tool name
        if name not in self._allowed_tools:
            raise ValueError(f"Tool not allowed: {name}")
        
        # Validate arguments
        schema = self._tool_schemas[name]
        validated = self._validate_against_schema(arguments, schema)
        
        # Sanitize inputs
        sanitized = self._sanitize_inputs(validated)
        
        return await self._execute_tool(name, sanitized)

MCP vs Alternative Approaches

Approach Pros Cons Use Case
MCP Standardized, composable, secure Still maturing Modern AI apps
RAG Well-understood, flexible Manual integration Document search
Function Calling Direct LLM support Provider-specific Simple tools
Custom APIs Full control No standardization Legacy systems
Plugins Rich ecosystem Security risks Consumer apps

The Future of MCP

Coming Soon:

  • Streaming support for real-time data
  • Multi-modal resources (images, video)
  • Federated MCP servers (compose multiple servers)
  • MCP marketplaces (discover and install servers)
  • Browser-native MCP clients

Ecosystem Growth:

  • Every SaaS will offer an MCP server
  • Desktop apps exposing data via MCP
  • IDE plugins for MCP development
  • MCP server registries and discovery

Getting Started

1. Install MCP SDK:

1

pip install anthropic-mcp

2. Run Example Server:

1
2
3
4
5
6
7
8
9
10
11
12

from mcp.server import Server

server = Server(
    name="hello-mcp",
    description="A simple MCP server"
)

@server.resource("greeting://")
def get_greeting():
    return "Hello from MCP!"

server.run()

3. Connect Client:

1
2
3
4
5

from mcp import Client

client = Client("http://localhost:8000")
greeting = client.read_resource("greeting://")
print(greeting)  # "Hello from MCP!"

Conclusion

The Model Context Protocol represents a paradigm shift: AI models become first-class citizens in application architecture. Instead of apps calling AI, AI calls apps—discovering resources, using tools, and composing capabilities dynamically.

For AI engineers, MCP means less time building custom integrations and more time building intelligent systems. For application developers, MCP means AI-native APIs that work across all models and platforms.

The protocol is young, but the momentum is undeniable. Every major AI platform is adopting MCP. The question isn’t whether to learn it, but how quickly you can start building with it.

Suggested Reading

AI, Backend
mcp model-context-protocol ai-applications llm anthropic web-development
This post is licensed under CC BY 4.0 by the author.