### The problem MCP solves You have built an AI agent. It can reason about incidents, suggest fixes, and answer questions about your infrastructure. But every time it needs real data — current pod status, live metrics, the latest log lines — it cannot get them on its own. You have to copy-paste data into the conversation manually. The agent is smart but blind. The traditional solution was **function calling** — you define tools directly in your API call and the model generates JSON that your code executes. This works for a single application. It breaks down when you want multiple AI tools to share the same integrations, or when you want to add a new tool without modifying every application that uses it. **MCP (Model Context Protocol)** is an open standard created by Anthropic that solves this differently. Instead of defining tools inside each application, you build a standalone **MCP server** — a small service that exposes tools over a standard protocol. Any AI application that speaks MCP can connect to your server and use its tools immediately, without you changing the application code. Think of it like this: before USB, every device needed its own proprietary connector. USB standardised the interface so any device could plug into any computer. MCP does the same for AI tools. Without MCP With MCP ────────────── ────────────── App 1 has its own kubectl tools MCP Server: ops-tools App 2 has its own kubectl tools | App 3 has its own kubectl tools exposes: kubectl, prometheus, runbooks | | Duplicated code everywhere Claude, GPT, any MCP client connects Each app maintains its own tools One server, shared by all apps ### What MCP actually is under the hood MCP is a protocol — a defined way for a client (an AI application) and a server (your tool provider) to communicate. The server exposes three types of things: * **Tools** — functions the AI can call to take actions or read data. `get_pod_status`, `query_prometheus`, `search_runbooks`. * **Resources** — data the AI can read directly, like file contents or database records. * **Prompts** — pre-built prompt templates the AI can use. For ops use cases, tools are what matter most. A tool has a name, a description the AI reads to understand when to use it, and a defined set of parameters. When an AI agent needs information, it reads the tool descriptions, picks the right tool, generates the parameters, and sends a tool call to your MCP server. Your server runs the actual code and sends back the result. AI Agent Your MCP Server | | |-- "I need pod status" --> | | |-- runs: kubectl get pods |<-- {pod list data} ----- | | | |-- uses data to answer --> | The AI never directly runs `kubectl`. It calls your MCP server, which runs `kubectl` and returns structured data. This is the safety boundary. ---
### Why the comparison matters If you have worked with the OpenAI or Anthropic API before, you have used function calling — defining tools directly inside your API call. MCP is a different approach. Understanding the difference helps you decide when to use each and explains clearly what MCP adds. | | Function Calling | MCP | |--|-----------------|-----| | Where tools are defined | Inside each app's code | In a standalone server | | Reuse across apps | Copy-paste or shared library | Connect any app to the same server | | Adding a new tool | Change every app that needs it | Add once to the server | | Protocol | Each API has its own format | Standard protocol — works with any MCP client | | Who can use it | Only apps you control | Any MCP-compatible client | | Coupling | Tight — app and tools bundled together | Loose — server and clients independent | **When to use function calling:** You have one application, one team, simple tools, and no need to share them. Function calling is simpler to set up. **When to use MCP:** You want multiple AI applications to share the same tools, you want to add tools without modifying application code, or you want to build a reusable ops tool library your team can connect anything to. For an AIOps platform team at a company like Razorpay or Zerodha, MCP makes sense — you build one ops server and every AI tool the team builds can use it. Each new incident response tool, each new chatbot, each new automated workflow — they all share the same verified, secured, production-tested ops integrations. ### The complete architecture Here is how all the pieces connect in a production AIOps setup. The MCP server sits in the middle, between the AI clients and your actual infrastructure: Claude Desktop / Claude API / Custom Agent | MCP Protocol | v ┌─────────────────────┐ │ Ops MCP Server │ │ ops_mcp_server.py │ └──────┬──────┬───────┘ | | | v v v Kubernetes Prometheus Runbooks (kubectl) (HTTP API) (files/RAG) The MCP server is the only component that has direct access to your infrastructure. The AI client never touches Kubernetes, Prometheus, or your runbook files directly. All access goes through the server, where you control permissions, logging, and guardrails. ---
### What tools an ops MCP server should expose Before writing code, design your tool surface. Too many tools confuses the AI — it cannot choose correctly when it has 50 options. Too few tools and the agent cannot accomplish its goals. A good ops MCP server exposes tools in three categories: **Read tools** — safe, no side effects, auto-approved: * Get pod status and logs * Query Prometheus metrics * Search runbooks and documentation * Get recent deployment history * Describe Kubernetes resources **Action tools** — change state but reversible, require logging: * Restart a deployment * Scale a deployment up or down * Roll back a deployment to previous version **Escalation tools** — notify humans, never execute automatically: * Create a PagerDuty alert * Post to a Slack channel * Create a JIRA ticket > ⚠️ **Security:** Never expose tools that delete resources, modify secrets, or execute arbitrary shell commands. > 💡 **Tip:** In this module, the Prometheus tool exposes fixed queries for specific metrics like error rate and latency. In production, many teams expose a single `run_promql(query)` tool that accepts any PromQL query — but always validate the query before running it to prevent injection or accidental expensive queries. For learning, specific tools are clearer. An MCP server is a power multiplier for an AI agent — the blast radius of a mistake is larger than with a human. Design your tool surface assuming the AI will occasionally make wrong decisions. ### Tool design principles Every tool should: * Do one specific thing — not "manage kubernetes" but "get pod status in namespace" * Have a clear description the AI can read to understand when to use it * Return structured data the AI can reason about — JSON, not raw text * Fail safely — if something goes wrong, return an error message, never crash A well-described tool means the AI picks it correctly. A poorly described tool means the AI either never uses it or uses it for the wrong purpose. Good tool description: ```text get_pod_status: Get the current status of all pods in a Kubernetes namespace, including phase (Running/Pending/Failed), restart count, and readiness. Use this when investigating service health or after a deployment. Parameters: namespace (string) - the Kubernetes namespace to check ``` Bad tool description: ```text get_pods: Gets pods Parameters: ns (string) ``` ---
### Installing the MCP SDK ```bash ## Install the official MCP Python SDK from Anthropic pip install mcp ## Install dependencies for our ops tools pip install kubernetes prometheus-api-client requests ``` ### The minimal MCP server structure Every MCP server needs three things: an MCP server instance, tool definitions with descriptions, and a transport to communicate over. Here is the absolute minimum: ```python # minimal_mcp_server.py # The simplest possible MCP server — one tool, runs and exits from mcp.server import Server from mcp.server.stdio import stdio_server from mcp import types import asyncio # Create the server instance — this is the MCP application # The name appears in client logs and helps identify which server is running server = Server("minimal-ops-server") # @server.list_tools() tells the MCP framework what tools this server offers # The client calls this first to discover available tools @server.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="say_hello", # Description is what the AI reads to decide when to use this tool # Write it like documentation for a smart engineer, not a user manual description="Returns a greeting. Use this to test if the MCP server is running.", inputSchema={ "type": "object", "properties": { "name": { "type": "string", "description": "The name to greet" } }, "required": ["name"] } ) ] # @server.call_tool() handles the actual execution when the AI calls a tool # tool_name: which tool was called # arguments: the parameters the AI provided @server.call_tool() async def call_tool(tool_name: str, arguments: dict) -> list[types.TextContent]: if tool_name == "say_hello": name = arguments.get("name", "world") return [types.TextContent(type="text", text=f"Hello, {name}! MCP server is working.")] # Always handle unknown tool names — the AI might call something that does not exist return [types.TextContent(type="text", text=f"Unknown tool: {tool_name}")] # Run the server over stdio transport # stdio means the server communicates via standard input/output # Claude Desktop and most MCP clients use stdio by default async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) if __name__ == "__main__": asyncio.run(main()) ``` > **Note:** MCP servers use `async/await` because they handle multiple concurrent requests. `asyncio` is Python's standard library for asynchronous programming. If you have not used async Python before, the key thing to know is: `async def` defines a function that can be paused while waiting for I/O, and `await` is used when calling another async function. The `asyncio.run()` at the bottom starts the event loop that runs everything. ### Building a real ops MCP server Now a real server with tools for Kubernetes, Prometheus, and runbook search: ```python # ops_mcp_server.py # A production-ready MCP server for AIOps — gives an AI agent # safe, structured access to your infrastructure import asyncio import json import subprocess from typing import Any import requests from mcp.server import Server from mcp.server.stdio import stdio_server from mcp import types # Kubernetes client for the cluster API from kubernetes import client, config # ─── Server Setup ──────────────────────────────────────────────────────────── server = Server("ops-mcp-server") # Load Kubernetes config at startup — same as kubectl uses # If this fails, Kubernetes tools will return errors but the server keeps running try: config.load_kube_config() core_v1 = client.CoreV1Api() apps_v1 = client.AppsV1Api() K8S_AVAILABLE = True except Exception as e: print(f"Warning: Kubernetes not available: {e}") K8S_AVAILABLE = False # Prometheus URL — set this to your actual Prometheus endpoint PROMETHEUS_URL = "http://localhost:9090" # Path to your runbooks directory — the MCP server will search these files RUNBOOKS_DIR = "./runbooks" # ─── Tool Registry ──────────────────────────────────────────────────────────── @server.list_tools() async def list_tools() -> list[types.Tool]: """ Register all available tools with their descriptions. The AI reads these descriptions to decide which tool to call. Write descriptions that are precise about when and how to use each tool. """ return [ # READ TOOLS — safe, no side effects types.Tool( name="get_pod_status", description=( "Get the current status of all pods in a Kubernetes namespace. " "Returns pod name, phase (Running/Pending/Failed/CrashLoopBackOff), " "restart count, and readiness. Use this when investigating service " "health, after a deployment, or when alerts indicate pod issues." ), inputSchema={ "type": "object", "properties": { "namespace": { "type": "string", "description": "Kubernetes namespace to check, e.g. 'payment' or 'production'" } }, "required": ["namespace"] } ), types.Tool( name="get_pod_logs", description=( "Get recent error logs from pods of a specific service. " "Returns the most recent error and exception log lines. " "Use this when investigating why a service is failing or " "when you need to see what errors are being thrown." ), inputSchema={ "type": "object", "properties": { "namespace": { "type": "string", "description": "Kubernetes namespace" }, "service": { "type": "string", "description": "Service name (matches the 'app' label on pods)" }, "since_minutes": { "type": "integer", "description": "How many minutes of logs to retrieve. Default: 15", "default": 15 } }, "required": ["namespace", "service"] } ), types.Tool( name="query_prometheus", description=( "Query Prometheus for current metrics. Returns current error rate " "and p99 latency for a service. Use this to check if a service " "is currently experiencing elevated errors or high latency." ), inputSchema={ "type": "object", "properties": { "service": { "type": "string", "description": "Service name as it appears in Prometheus labels" } }, "required": ["service"] } ), types.Tool( name="search_runbooks", description=( "Search internal runbooks and operational documentation for " "procedures related to a specific problem. Returns the most " "relevant runbook sections. Use this when you need to find " "the correct procedure for handling a specific alert or failure." ), inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The problem or alert to search for, e.g. 'redis memory high' or 'pod crashloopbackoff'" } }, "required": ["query"] } ), # ACTION TOOLS — change state, reversible, always logged types.Tool( name="restart_deployment", description=( "Restart all pods in a Kubernetes deployment by triggering a " "rolling restart. This is safe — pods are replaced one at a time. " "Use this when pods are in a bad state and need a clean restart. " "Always check pod status after restarting to confirm recovery." ), inputSchema={ "type": "object", "properties": { "namespace": { "type": "string", "description": "Kubernetes namespace" }, "deployment": { "type": "string", "description": "Deployment name to restart" } }, "required": ["namespace", "deployment"] } ), ] # ─── Tool Implementations ───────────────────────────────────────────────────── @server.call_tool() async def call_tool(tool_name: str, arguments: dict) -> list[types.TextContent]: """ Route tool calls to the correct implementation. All tools return TextContent with JSON-formatted results. Errors are returned as JSON too — never crash, always return a result. """ if tool_name == "get_pod_status": return await tool_get_pod_status(arguments) elif tool_name == "get_pod_logs": return await tool_get_pod_logs(arguments) elif tool_name == "query_prometheus": return await tool_query_prometheus(arguments) elif tool_name == "search_runbooks": return await tool_search_runbooks(arguments) elif tool_name == "restart_deployment": return await tool_restart_deployment(arguments) else: return [types.TextContent(type="text", text=json.dumps({"error": f"Unknown tool: {tool_name}"}))] async def tool_get_pod_status(args: dict) -> list[types.TextContent]: """Get pod status for all pods in a namespace.""" namespace = args["namespace"] if not K8S_AVAILABLE: return [types.TextContent(type="text", text=json.dumps({"error": "Kubernetes not configured"}))] try: pods = core_v1.list_namespaced_pod(namespace) result = [] for pod in pods.items: # Extract restart count from container status restart_count = 0 if pod.status.container_statuses: restart_count = sum(cs.restart_count for cs in pod.status.container_statuses) result.append({ "name": pod.metadata.name, "phase": pod.status.phase, "ready": all(cs.ready for cs in (pod.status.container_statuses or [])), "restarts": restart_count, "node": pod.spec.node_name }) return [types.TextContent(type="text", text=json.dumps(result, indent=2))] except Exception as e: return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))] async def tool_get_pod_logs(args: dict) -> list[types.TextContent]: """Get error logs from all pods of a service.""" namespace = args["namespace"] service = args["service"] since_minutes = args.get("since_minutes", 15) try: # Use subprocess to run kubectl — keeps the implementation simple # and uses the same auth as the engineer's terminal result = subprocess.run([ "kubectl", "logs", "-n", namespace, "-l", f"app={service}", "--since", f"{since_minutes}m", "--timestamps=true" ], capture_output=True, text=True, timeout=30) logs = result.stdout # Filter to error lines only — reduces noise for the AI error_lines = [ line for line in logs.split('\n') if any(w in line.lower() for w in ['error', 'exception', 'fatal', 'timeout', 'refused', 'exhausted']) ] response = { "service": service, "namespace": namespace, "since_minutes": since_minutes, "error_log_count": len(error_lines), "error_logs": error_lines[:30] # cap at 30 lines to stay within context limits } return [types.TextContent(type="text", text=json.dumps(response, indent=2))] except subprocess.TimeoutExpired: return [types.TextContent(type="text", text=json.dumps({"error": "kubectl timed out"}))] except Exception as e: return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))] async def tool_query_prometheus(args: dict) -> list[types.TextContent]: """Query current error rate and latency from Prometheus.""" service = args["service"] try: def query(promql: str) -> float: """Run a single PromQL query and return the scalar result.""" resp = requests.get( f"{PROMETHEUS_URL}/api/v1/query", params={"query": promql}, timeout=10 ) data = resp.json() if data["data"]["result"]: return float(data["data"]["result"][0]["value"][1]) return 0.0 result = { "service": service, # Errors per second over the last 5 minutes "error_rate_per_sec": query( f'sum(rate(http_requests_total{{service="{service}",status=~"5.."}}[5m]))' ), # 99th percentile latency in milliseconds "p99_latency_ms": query( f'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket{{service="{service}"}}[5m])) by (le))' ) * 1000, } return [types.TextContent(type="text", text=json.dumps(result, indent=2))] except Exception as e: return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))] async def tool_search_runbooks(args: dict) -> list[types.TextContent]: """Search runbook files for content matching the query.""" import os query = args["query"].lower() results = [] try: # Walk the runbooks directory and search each markdown file for filename in os.listdir(RUNBOOKS_DIR): if not filename.endswith(".md"): continue filepath = os.path.join(RUNBOOKS_DIR, filename) with open(filepath) as f: content = f.read() # Simple keyword search — for production use the RAG pipeline instead # This is a fallback for when the vector database is not available if any(word in content.lower() for word in query.split()): # Return the first 500 characters of the matching file results.append({ "source": filename, "preview": content[:500] }) if not results: return [types.TextContent(type="text", text=json.dumps({"message": "No runbooks found for this query", "query": query}))] return [types.TextContent(type="text", text=json.dumps(results, indent=2))] except Exception as e: return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))] async def tool_restart_deployment(args: dict) -> list[types.TextContent]: """Restart a Kubernetes deployment with a rolling restart.""" namespace = args["namespace"] deployment = args["deployment"] # Log every action tool call — this is your audit trail print(f"[ACTION] restart_deployment: {deployment} in {namespace}", flush=True) try: result = subprocess.run([ "kubectl", "rollout", "restart", f"deployment/{deployment}", "-n", namespace ], capture_output=True, text=True, timeout=30) if result.returncode == 0: return [types.TextContent(type="text", text=json.dumps({ "status": "success", "message": f"Rolling restart initiated for {deployment} in {namespace}", "next_step": "Use get_pod_status to monitor the rollout progress" }))] else: return [types.TextContent(type="text", text=json.dumps({ "status": "error", "message": result.stderr }))] except Exception as e: return [types.TextContent(type="text", text=json.dumps({"error": str(e)}))] # ─── Server Entry Point ─────────────────────────────────────────────────────── async def main(): """Start the MCP server over stdio transport.""" async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) if __name__ == "__main__": asyncio.run(main()) ``` ---
### Claude Desktop configuration Claude Desktop supports MCP servers natively. You configure them in a JSON file and they become available in every Claude conversation. Find the config file: * Mac: `~/Library/Application Support/Claude/claude_desktop_config.json` * Windows: `%APPDATA%\Claude\claude_desktop_config.json` Add your server: ```json { "mcpServers": { "ops-tools": { "command": "python3", "args": ["/path/to/your/ops_mcp_server.py"], "env": { "PROMETHEUS_URL": "http://localhost:9090", "RUNBOOKS_DIR": "/path/to/your/runbooks" } } } } ``` Restart Claude Desktop. Your tools appear automatically — Claude can now call them in any conversation. > 📌 **Remember:** The `command` and `args` run your server as a subprocess. Claude Desktop launches it and communicates via stdio. The server must be runnable from the terminal — test it with `python3 ops_mcp_server.py` first. ### Using your server with Claude via API For programmatic use — in your AIOps agent code: ```python # use_mcp_server.py # Using your MCP server from Python via the Claude API # This is how you build an AIOps agent that uses your ops tools import anthropic import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def run_ops_agent(user_question: str): """ Run an AIOps agent that can use your MCP server tools. The agent reads tool descriptions and decides which ones to call. """ # Connect to your MCP server as a subprocess # The server starts when this context manager opens server_params = StdioServerParameters( command="python3", args=["ops_mcp_server.py"] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Discover available tools from the server await session.initialize() tools_response = await session.list_tools() available_tools = tools_response.tools print(f"Connected to MCP server. {len(available_tools)} tools available.") # Convert MCP tool format to Anthropic API tool format # The Claude API expects tools in a specific schema format anthropic_tools = [ { "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema } for tool in available_tools ] client = anthropic.Anthropic() messages = [{"role": "user", "content": user_question}] print(f"\nQuestion: {user_question}\n") # Agent loop — runs until Claude stops calling tools while True: response = client.messages.create( model="claude-opus-4-6", max_tokens=2048, tools=anthropic_tools, messages=messages ) # Claude wants to call a tool if response.stop_reason == "tool_use": # Find the tool use block in the response tool_use = next(b for b in response.content if b.type == "tool_use") print(f"Calling tool: {tool_use.name}") print(f"Parameters: {tool_use.input}") # Execute the tool call on our MCP server tool_result = await session.call_tool(tool_use.name, tool_use.input) result_text = tool_result.content[0].text if tool_result.content else "{}" print(f"Result: {result_text[:200]}...\n" if len(result_text) > 200 else f"Result: {result_text}\n") # Add the tool call and result to conversation history # The agent uses this context to decide what to do next messages.append({"role": "assistant", "content": response.content}) messages.append({ "role": "user", "content": [{ "type": "tool_result", "tool_use_id": tool_use.id, "content": result_text }] }) # Claude has a final answer else: final = next(b for b in response.content if hasattr(b, "text")) print(f"Agent: {final.text}") break # Run the agent asyncio.run(run_ops_agent( "The payment service is throwing errors. Check pod status and recent logs in the payment namespace." )) ``` ### What the agent output looks like ```text Connected to MCP server. 5 tools available. Question: The payment service is throwing errors. Check pod status and recent logs in the payment namespace. Calling tool: get_pod_status Parameters: {'namespace': 'payment'} Result: [{"name": "payment-api-7d9b4c8f6-xkp2m", "phase": "Running", "ready": false, "restarts": 8, ...} Calling tool: get_pod_logs Parameters: {'namespace': 'payment', 'service': 'payment-api', 'since_minutes': 15} Result: {"error_log_count": 23, "error_logs": ["ERROR: connection pool exhausted... Agent: Based on the data I retrieved: Pod payment-api-7d9b4c8f6-xkp2m has restarted 8 times and is not ready. The logs show repeated "connection pool exhausted" errors starting 12 minutes ago. Most likely root cause: The Redis connection pool is exhausted. This typically happens when pool_size is too low or when connections are not being released properly. Recommended next steps: 1. Check the Redis connection pool configuration in the deployment config 2. Look for recent config changes: kubectl rollout history deployment/payment-api -n payment 3. If pool_size was reduced recently, restore it and run: restart_deployment ``` The agent checked two tools, combined the results, and gave a structured diagnosis — all without any manual data collection. ---
### Why approval matters in production The MCP server we built allows the AI to call `restart_deployment` directly. For a learning environment this is fine. For production systems handling real traffic at Zerodha or Razorpay, this is risky. The problem: AI agents occasionally make incorrect decisions — especially when alert data is ambiguous or when multiple possible causes exist. A restart triggered on the wrong service during peak trading hours is worse than no action at all. The solution is an approval gate between the AI's decision and the actual execution: AI decides: restart payment-api in production | v Approval Required Send to Slack channel Wait for engineer response | ┌---------┴---------┐ | | v v Approved Denied | | v v Execute restart Cancel + log reason This pattern keeps automation speed (the AI prepares and proposes the action) while keeping humans accountable for production changes. ### Adding a Slack approval gate to restart_deployment ```python # approval_gate.py # A simple Slack-based approval workflow for high-risk MCP tool calls # Requires: pip install slack-sdk import asyncio import time import uuid import requests SLACK_BOT_TOKEN = "xoxb-your-bot-token" # set in environment APPROVAL_CHANNEL = "#ops-approvals" # your on-call channel APPROVAL_TIMEOUT_SECONDS = 300 # 5 minutes to approve or deny # In-memory store for pending approvals # In production, use Redis or a database so approvals survive restarts pending_approvals: dict = {} def request_approval(action: str, details: dict) -> str: """ Post an approval request to Slack and return an approval ID. The AI calls this instead of executing the action directly. Returns the approval_id to track the request. """ approval_id = str(uuid.uuid4())[:8] # short ID for easy reference # Store the pending request pending_approvals[approval_id] = { "action": action, "details": details, "status": "pending", "requested_at": time.time() } # Build the Slack message with approve/deny buttons message = { "channel": APPROVAL_CHANNEL, "text": f"*AI Action Approval Required* (ID: `{approval_id}`)", "blocks": [ { "type": "section", "text": { "type": "mrkdwn", "text": ( f"*Action:* `{action}` " f"*Details:* {details} " f"*Approval ID:* `{approval_id}` " f"*Expires in:* 5 minutes" ) } }, { "type": "actions", "elements": [ { "type": "button", "text": {"type": "plain_text", "text": "Approve"}, "style": "primary", # action_id carries the approval_id so the webhook knows which request "action_id": f"approve_{approval_id}", "value": approval_id }, { "type": "button", "text": {"type": "plain_text", "text": "Deny"}, "style": "danger", "action_id": f"deny_{approval_id}", "value": approval_id } ] } ] } # Post to Slack requests.post( "https://slack.com/api/chat.postMessage", headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"}, json=message ) return approval_id async def wait_for_approval(approval_id: str, timeout: int = APPROVAL_TIMEOUT_SECONDS) -> dict: """ Poll the pending_approvals dict until the engineer approves, denies, or timeout. In production, use a webhook endpoint instead of polling. """ start = time.time() while time.time() - start < timeout: approval = pending_approvals.get(approval_id, {}) if approval.get("status") == "approved": return {"approved": True, "by": approval.get("approved_by", "unknown")} if approval.get("status") == "denied": return {"approved": False, "reason": approval.get("deny_reason", "Denied by engineer")} # Check every 5 seconds — avoid hammering the store await asyncio.sleep(5) return {"approved": False, "reason": "Approval timed out after 5 minutes"} # Updated restart_deployment tool with approval gate async def tool_restart_deployment_with_approval(args: dict): """ Restart deployment — requires human approval via Slack before executing. Returns immediately with a pending message, then waits for approval. """ from mcp import types import json import datetime namespace = args["namespace"] deployment = args["deployment"] # Hard block on protected namespaces — no approval can override this PROTECTED = ["kube-system", "kube-public", "kube-node-lease"] if namespace in PROTECTED: return [types.TextContent(type="text", text=json.dumps({ "error": f"Namespace '{namespace}' is protected — cannot be restarted" }))] # Request approval approval_id = request_approval( action="restart_deployment", details={"deployment": deployment, "namespace": namespace} ) print(f"[{datetime.datetime.utcnow().isoformat()}] Approval requested: " f"{deployment}/{namespace} (ID: {approval_id})", flush=True) # Wait for engineer response result = await wait_for_approval(approval_id) if result["approved"]: # Execute the restart import subprocess subprocess.run([ "kubectl", "rollout", "restart", f"deployment/{deployment}", "-n", namespace ], capture_output=True) print(f"[{datetime.datetime.utcnow().isoformat()}] Approved and executed by {result['by']}") return [types.TextContent(type="text", text=json.dumps({ "status": "executed", "approved_by": result["by"], "message": f"Rolling restart initiated for {deployment} in {namespace}" }))] else: return [types.TextContent(type="text", text=json.dumps({ "status": "denied", "reason": result["reason"], "message": "Action was denied — no changes made" }))] ``` > 📌 **Remember:** The approval pattern applies to any action tool that changes production state. Restart, scale, rollback — all should go through a human gate. Read-only tools like get_pod_status and query_prometheus never need approval. Design your tool surface so the boundary between read and action is always clear. ---
The problem MCP solves You have built an AI agent. It can reason about incidents, suggest fixes, and answer questions ab...
Why the comparison matters If you have worked with the OpenAI or Anthropic API before, you have used function calling — ...
What tools an ops MCP server should expose Before writing code, design your tool surface. Too many tools confuses the AI...
Installing the MCP SDK The minimal MCP server structure Every MCP server needs three things: an MCP server instance, too...
Claude Desktop configuration Claude Desktop supports MCP servers natively. You configure them in a JSON file and they be...
Why approval matters in production The MCP server we built allows the AI to call restartdeployment directly. For a learn...
Why keyword search is not enough in production The searchrunbooks tool in our server uses simple keyword matching — it c...
What you must never expose as MCP tools MCP tools give an AI agent real power over your infrastructure. The wrong tool e...
Connecting all the tools into one workflow Now that you have a working MCP server, here is what the full incident invest...
Test tools directly before connecting to Claude Before connecting to Claude Desktop or the API, test each tool directly ...
MCP server quick reference Concept What it is Server("name") Creates the MCP server instance @server.listtools() Registe...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.