Model Context Protocol (MCP)

# MCP Architecture Overview

# SERVER: Exposes tools and resources
class MCPServer:
    name: "github-server"
    version: "1.0.0"

    # Tools the server provides
    tools: [
        { name: "search_repos", description: "Search GitHub repositories" },
        { name: "get_file", description: "Get file contents from repo" },
        { name: "create_issue", description: "Create a new issue" }
    ]

    # Resources the server exposes
    resources: [
        { uri: "github://repos/{owner}/{repo}", description: "Repository info" },
        { uri: "github://issues/{owner}/{repo}", description: "Issue list" }
    ]

    # Handle tool invocation
    function handleToolCall(toolName, arguments):
        switch toolName:
            case "search_repos":
                return githubAPI.searchRepos(arguments.query)
            case "get_file":
                return githubAPI.getFile(arguments.repo, arguments.path)
            # ...

# CLIENT: Connects to servers and routes requests
class MCPClient:
    servers: Map<string, MCPServer>

    function connect(serverConfig):
        server = spawn(serverConfig.command, serverConfig.args)
        capabilities = server.initialize()
        servers.set(serverConfig.name, server)
        return capabilities

    function callTool(serverName, toolName, arguments):
        server = servers.get(serverName)
        return server.tools.call(toolName, arguments)

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent, Resource
import mcp.server.stdio

# Create an MCP server
server = Server("github-server")

# Define available tools
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_repos",
            description="Search GitHub repositories by query",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query"
                    },
                    "limit": {
                        "type": "integer",
                        "default": 10
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="get_file",
            description="Get file contents from a repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "path": {"type": "string"}
                },
                "required": ["owner", "repo", "path"]
            }
        ),
        Tool(
            name="create_issue",
            description="Create a new issue in a repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "title": {"type": "string"},
                    "body": {"type": "string"}
                },
                "required": ["owner", "repo", "title"]
            }
        )
    ]

# Handle tool calls
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_repos":
        results = await github_api.search_repos(arguments["query"])
        return [TextContent(type="text", text=json.dumps(results))]

    elif name == "get_file":
        content = await github_api.get_file(
            arguments["owner"],
            arguments["repo"],
            arguments["path"]
        )
        return [TextContent(type="text", text=content)]

    elif name == "create_issue":
        issue = await github_api.create_issue(
            arguments["owner"],
            arguments["repo"],
            arguments["title"],
            arguments.get("body", "")
        )
        return [TextContent(type="text", text=f"Created issue #{issue.number}")]

    raise ValueError(f"Unknown tool: {name}")

# Define available resources
@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="github://repos/{owner}/{repo}",
            name="Repository Information",
            description="Get metadata about a GitHub repository",
            mimeType="application/json"
        )
    ]

# Run the server
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(
            read,
            write,
            InitializationOptions(
                server_name="github-server",
                server_version="1.0.0"
            )
        )

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

using Microsoft.Extensions.AI;
using ModelContextProtocol;
using ModelContextProtocol.Server;

// Define an MCP server
public class GitHubMcpServer : McpServer
{
    private readonly IGitHubClient _github;

    public GitHubMcpServer(IGitHubClient github)
    {
        _github = github;
    }

    public override ServerInfo GetServerInfo() => new()
    {
        Name = "github-server",
        Version = "1.0.0"
    };

    public override IEnumerable<Tool> ListTools()
    {
        yield return new Tool
        {
            Name = "search_repos",
            Description = "Search GitHub repositories by query",
            InputSchema = new JsonSchema
            {
                Type = "object",
                Properties = new Dictionary<string, JsonSchema>
                {
                    ["query"] = new() { Type = "string", Description = "Search query" },
                    ["limit"] = new() { Type = "integer", Default = 10 }
                },
                Required = new[] { "query" }
            }
        };

        yield return new Tool
        {
            Name = "get_file",
            Description = "Get file contents from a repository",
            InputSchema = new JsonSchema
            {
                Type = "object",
                Properties = new Dictionary<string, JsonSchema>
                {
                    ["owner"] = new() { Type = "string" },
                    ["repo"] = new() { Type = "string" },
                    ["path"] = new() { Type = "string" }
                },
                Required = new[] { "owner", "repo", "path" }
            }
        };
    }

    public override async Task<ToolResult> CallToolAsync(
        string name,
        JsonElement arguments,
        CancellationToken ct = default)
    {
        return name switch
        {
            "search_repos" => await SearchReposAsync(arguments, ct),
            "get_file" => await GetFileAsync(arguments, ct),
            "create_issue" => await CreateIssueAsync(arguments, ct),
            _ => throw new McpException($"Unknown tool: {name}")
        };
    }

    private async Task<ToolResult> SearchReposAsync(
        JsonElement args,
        CancellationToken ct)
    {
        var query = args.GetProperty("query").GetString()!;
        var limit = args.TryGetProperty("limit", out var l) ? l.GetInt32() : 10;

        var results = await _github.SearchRepositoriesAsync(query, limit, ct);
        return ToolResult.Success(JsonSerializer.Serialize(results));
    }
}

// Run the server
var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddMcpServer<GitHubMcpServer>();

var host = builder.Build();
await host.RunAsync();

from mcp.server import Server
from mcp.types import Resource, ResourceContents, TextResourceContents

server = Server("data-server")

# List available resources
@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="data://users",
            name="User Directory",
            description="List of all users in the system",
            mimeType="application/json"
        ),
        Resource(
            uri="data://users/{user_id}",
            name="User Profile",
            description="Detailed profile for a specific user",
            mimeType="application/json"
        ),
        Resource(
            uri="data://reports/{report_type}/{date}",
            name="Reports",
            description="Generated reports by type and date",
            mimeType="application/json"
        )
    ]

# Read a specific resource
@server.read_resource()
async def read_resource(uri: str) -> ResourceContents:
    # Parse URI to extract resource type and parameters
    parts = uri.replace("data://", "").split("/")

    if parts[0] == "users":
        if len(parts) == 1:
            # List all users
            users = await db.get_all_users()
            return TextResourceContents(
                uri=uri,
                mimeType="application/json",
                text=json.dumps(users)
            )
        else:
            # Get specific user
            user_id = parts[1]
            user = await db.get_user(user_id)
            return TextResourceContents(
                uri=uri,
                mimeType="application/json",
                text=json.dumps(user)
            )

    elif parts[0] == "reports":
        report_type = parts[1]
        date = parts[2]
        report = await generate_report(report_type, date)
        return TextResourceContents(
            uri=uri,
            mimeType="application/json",
            text=json.dumps(report)
        )

    raise ValueError(f"Unknown resource: {uri}")

# Subscribe to resource changes (optional)
@server.subscribe_resource()
async def subscribe_resource(uri: str):
    # Set up change notifications
    async def on_change():
        await server.notify_resource_updated(uri)

    await db.watch(uri, on_change)

public class DataMcpServer : McpServer
{
    private readonly IDatabase _db;

    public override IEnumerable<Resource> ListResources()
    {
        yield return new Resource
        {
            Uri = "data://users",
            Name = "User Directory",
            Description = "List of all users in the system",
            MimeType = "application/json"
        };

        yield return new Resource
        {
            Uri = "data://users/{user_id}",
            Name = "User Profile",
            Description = "Detailed profile for a specific user",
            MimeType = "application/json"
        };

        yield return new Resource
        {
            Uri = "data://reports/{report_type}/{date}",
            Name = "Reports",
            Description = "Generated reports by type and date",
            MimeType = "application/json"
        };
    }

    public override async Task<ResourceContents> ReadResourceAsync(
        string uri,
        CancellationToken ct = default)
    {
        var parts = uri.Replace("data://", "").Split('/');

        if (parts[0] == "users")
        {
            if (parts.Length == 1)
            {
                var users = await _db.GetAllUsersAsync(ct);
                return new TextResourceContents
                {
                    Uri = uri,
                    MimeType = "application/json",
                    Text = JsonSerializer.Serialize(users)
                };
            }
            else
            {
                var userId = parts[1];
                var user = await _db.GetUserAsync(userId, ct);
                return new TextResourceContents
                {
                    Uri = uri,
                    MimeType = "application/json",
                    Text = JsonSerializer.Serialize(user)
                };
            }
        }

        if (parts[0] == "reports")
        {
            var reportType = parts[1];
            var date = parts[2];
            var report = await GenerateReportAsync(reportType, date, ct);
            return new TextResourceContents
            {
                Uri = uri,
                MimeType = "application/json",
                Text = JsonSerializer.Serialize(report)
            };
        }

        throw new McpException($"Unknown resource: {uri}");
    }
}

# MCP Security Considerations

# 1. TRANSPORT ISOLATION
# Servers run as separate processes
# Communication via stdio (no network exposure)
server = spawn("mcp-server", stdin=PIPE, stdout=PIPE)

# 2. CAPABILITY NEGOTIATION
# Client declares what it supports
# Server only exposes agreed capabilities
capabilities = server.initialize({
    client: { name: "my-app", version: "1.0" },
    capabilities: {
        tools: { listChanged: true },
        resources: { subscribe: false },
        prompts: {}  # Not using prompts
    }
})

# 3. ARGUMENT VALIDATION
# Always validate before passing to tools
function validateToolCall(toolName, arguments):
    schema = getToolSchema(toolName)
    if not validate(arguments, schema):
        return Error("Invalid arguments")
    # Additional domain-specific validation
    if toolName == "delete_file" and arguments.path.startsWith("/"):
        return Error("Absolute paths not allowed")

# 4. RESOURCE SCOPING
# Limit what servers can access
server.config = {
    allowedPaths: ["/home/user/projects"],
    deniedPatterns: ["**/*.env", "**/secrets/*"],
    maxTokensPerRequest: 10000
}

# 5. AUDIT LOGGING
function callTool(server, tool, args):
    log({
        timestamp: now(),
        server: server.name,
        tool: tool,
        arguments: sanitize(args),  # Remove secrets
        user: getCurrentUser()
    })
    return server.callTool(tool, args)

from mcp.server import Server
from mcp.types import Tool
import os
from pathlib import Path

class SecureMcpServer(Server):
    def __init__(self, allowed_paths: list[str]):
        super().__init__("secure-server")
        self.allowed_paths = [Path(p).resolve() for p in allowed_paths]
        self.denied_patterns = ["*.env", "*.key", "secrets/*"]

    def validate_path(self, path: str) -> Path:
        """Ensure path is within allowed directories."""
        resolved = Path(path).resolve()

        # Check if within allowed paths
        is_allowed = any(
            resolved.is_relative_to(allowed)
            for allowed in self.allowed_paths
        )

        if not is_allowed:
            raise PermissionError(f"Path not allowed: {path}")

        # Check denied patterns
        for pattern in self.denied_patterns:
            if resolved.match(pattern):
                raise PermissionError(f"Access denied: {path}")

        return resolved

    @server.call_tool()
    async def call_tool(self, name: str, arguments: dict):
        # Validate all path arguments
        if "path" in arguments:
            arguments["path"] = str(self.validate_path(arguments["path"]))

        if "paths" in arguments:
            arguments["paths"] = [
                str(self.validate_path(p))
                for p in arguments["paths"]
            ]

        # Proceed with validated arguments
        return await self._execute_tool(name, arguments)

# Environment-based configuration
server = SecureMcpServer(
    allowed_paths=os.environ.get("MCP_ALLOWED_PATHS", "").split(":"),
)

public class SecureMcpServer : McpServer
{
    private readonly string[] _allowedPaths;
    private readonly string[] _deniedPatterns = { "*.env", "*.key", "secrets/*" };
    private readonly ILogger _logger;

    public SecureMcpServer(string[] allowedPaths, ILogger logger)
    {
        _allowedPaths = allowedPaths
            .Select(p => Path.GetFullPath(p))
            .ToArray();
        _logger = logger;
    }

    private string ValidatePath(string path)
    {
        var resolved = Path.GetFullPath(path);

        // Check if within allowed paths
        var isAllowed = _allowedPaths.Any(allowed =>
            resolved.StartsWith(allowed, StringComparison.OrdinalIgnoreCase));

        if (!isAllowed)
            throw new UnauthorizedAccessException($"Path not allowed: {path}");

        // Check denied patterns
        foreach (var pattern in _deniedPatterns)
        {
            if (FileSystemName.MatchesSimpleExpression(pattern, resolved))
                throw new UnauthorizedAccessException($"Access denied: {path}");
        }

        return resolved;
    }

    public override async Task<ToolResult> CallToolAsync(
        string name,
        JsonElement arguments,
        CancellationToken ct = default)
    {
        // Audit log
        _logger.LogInformation(
            "Tool call: {Tool} with args: {Args}",
            name,
            SanitizeForLog(arguments)
        );

        // Validate paths in arguments
        var validatedArgs = ValidateArguments(arguments);

        try
        {
            return await ExecuteToolAsync(name, validatedArgs, ct);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Tool {Tool} failed", name);
            throw;
        }
    }

    private string SanitizeForLog(JsonElement args)
    {
        // Remove sensitive fields from logs
        var doc = JsonDocument.Parse(args.GetRawText());
        var sanitized = new Dictionary<string, object>();

        foreach (var prop in doc.RootElement.EnumerateObject())
        {
            if (prop.Name.Contains("token", StringComparison.OrdinalIgnoreCase) ||
                prop.Name.Contains("password", StringComparison.OrdinalIgnoreCase) ||
                prop.Name.Contains("secret", StringComparison.OrdinalIgnoreCase))
            {
                sanitized[prop.Name] = "[REDACTED]";
            }
            else
            {
                sanitized[prop.Name] = prop.Value.ToString();
            }
        }

        return JsonSerializer.Serialize(sanitized);
    }
}

# MCP Apps: Interactive UI Components

# Tool returns UI metadata instead of plain text
function searchAnalytics(query, dateRange):
    results = database.query(query, dateRange)

    # Instead of returning text, return UI reference
    return {
        content: "Found {results.count} matching records",
        _meta: {
            ui: {
                resourceUri: "ui://analytics-dashboard",
                data: {
                    results: results,
                    chartType: "bar",
                    interactive: true
                }
            }
        }
    }

# UI Resource served via ui:// scheme
# Contains sandboxed HTML/JavaScript bundle
resource "ui://analytics-dashboard":
    type: "text/html"
    content: bundledDashboardApp
    permissions: ["display", "user-input"]

# Client renders UI in sandboxed iframe
# Bidirectional JSON-RPC for communication
client.onToolResult = (result) => {
    if result._meta?.ui:
        iframe = createSandboxedIframe(result._meta.ui.resourceUri)
        iframe.postMessage(result._meta.ui.data)
}

import { App } from "@modelcontextprotocol/ext-apps";

// Initialize MCP App client
const app = new App();
await app.connect();

// Handle tool results with UI components
app.ontoolresult = (result) => {
  if (result._meta?.ui) {
    // Render interactive visualization
    renderChart(result.data);
  }
};

// Call server tool that returns UI
const response = await app.callServerTool({
  name: "fetch_analytics",
  arguments: {
    metric: "revenue",
    period: "Q4"
  },
});

// Update model context based on user interaction
// (e.g., user clicks a data point in the chart)
await app.updateModelContext({
  content: [{
    type: "text",
    text: "User selected Q4 revenue breakdown by region"
  }],
});

// Server-side: Tool that returns UI metadata
server.tool("fetch_analytics", async (args) => {
  const data = await analyticsDB.query(args);

  return {
    content: [{
      type: "text",
      text: `Found ${data.length} records for ${args.metric}`
    }],
    _meta: {
      ui: {
        resourceUri: "ui://analytics-chart",
        data: {
          records: data,
          chartConfig: {
            type: "bar",
            interactive: true,
            drilldown: true
          }
        }
      }
    }
  };
});

// Register UI resource
server.resource("ui://analytics-chart", {
  mimeType: "text/html",
  read: async () => {
    return {
      contents: analyticsChartBundle, // Bundled HTML/JS
    };
  }
});

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

server = Server("analytics-server")

# Tool that returns interactive UI
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "analyze_data":
        results = await db.query(arguments["query"])

        return {
            "content": [
                TextContent(
                    type="text",
                    text=f"Analysis complete: {len(results)} records"
                )
            ],
            "_meta": {
                "ui": {
                    "resourceUri": "ui://data-explorer",
                    "data": {
                        "results": results,
                        "visualization": "interactive-table",
                        "allowExport": True
                    }
                }
            }
        }

# UI Resource definition
@server.read_resource()
async def read_resource(uri: str):
    if uri == "ui://data-explorer":
        # Return bundled HTML/JS application
        return {
            "uri": uri,
            "mimeType": "text/html",
            "text": load_ui_bundle("data-explorer.html")
        }

# List available UI resources
@server.list_resources()
async def list_resources():
    return [
        Resource(
            uri="ui://data-explorer",
            name="Interactive Data Explorer",
            description="Explore and visualize query results",
            mimeType="text/html"
        ),
        Resource(
            uri="ui://config-wizard",
            name="Configuration Wizard",
            description="Step-by-step configuration interface",
            mimeType="text/html"
        )
    ]