Model Context Protocol (MCP) is a standard for how AI models connect to external tools, data sources, and applications in a consistent way.
At a high level, MCP solves a common problem: How does an AI model reliably access and use outside information or services while keeping things structured, secure and predictable.
MCP defines a unified interface between and AI model and "context providers".
- Model = AI Model (Ex: ChatGPT)
- Context Providers = tools, APIs, databases, files, apps etc.
Instead of every tool having its own custom integration, MCP standardizes how the model:
- Discover available tools
- Requests data or actions
- Received structured responses
BTW - MCP was introduced by Anthropic in November 2024.
Example:
Let us discuss one example to understand MCP. Assume, we have below 5 data sources in our project.
- Confluence
- GitHub
- Websites
- Databases
- Filesystems
For each of the above data sources, we need to write a wrapper to extract the data, and if we have hundred's and thousands of data sources then we have to write those many wrappers. Hence to resolve this dependency issue, MCP came into picture.
Anthropic created a common interface called MCP and the sources will connect to this MCP. You can add 'n' of data sources to our project, still just an MCP is enough to handle all these data sources.
How MCP accepts these data sources ?
Tool - You are going to create a Tool with respect to the source system and register these tools in MCP server. MCP server is having the tools related to confluence page, GitHub, SPO, websites, databases and a fie system.
Responsibility of Agentic AI developer is creating these Tool and integrate those tools with MCP server.
When a user ask a question, it won't directly hit the MCP server. It will hit the MCP client(LLM) first, it understand the user query and redirect the request to MCP server, then MCP will call respective tool based on the user query.
This is the complete manual flow of a MCP server.
MCP Vs RAG :
Please go through above image to understand the difference between MCP & RAG. MCP is not a replacement to RAG. We have use cases where we nee both in the same project.
Production grade AI Agent system using MCP + RAG :
In real time, we may to deal with use cases which needs both MCP(to handle real time data which is dynamic in nature) + RAG (static data sources from vector DB's).
MCP Server contains 3 main components :
- Tool - @mcp.tool()
- Resources - @mcp.resources()
- Prompt - @mcp.prompt()
Note : To execute all the programs, we need to use MCP inspector and the command is as below. Hence I am not going to paste the output screenshots. Please try in your laptop, install required libraries(when you run the program, it will suggest what to install in the errors - please follow instructions accordingly)
- mcp dev <python_file_name.py>
- then it will open a UI where we can interact with mcp tools, resources and prompts.
Implementation of implementing a simple MCP Server :
# ────────────────────────────────────────────────────────────
# EXAMPLE 1 — HELLO WORLD (absolute basics)
# Concepts: FastMCP, @mcp.tool, plain return value
# ────────────────────────────────────────────────────────────
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Hello Server")
@mcp.tool()
def say_hello(name: str) -> str:
"""Say hello to someone."""
return f"Hello, {name}! Welcome to MCP."
@mcp.tool()
def add_numbers(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
if __name__ == "__main__":
print("=" * 55)
print("EXAMPLE 1 — Hello World")
print("=" * 55)
print(say_hello("Arun"))
print(add_numbers(10, 20))
print("\n" + "=" * 55)
Output :
Explanation :
- Need to import FastMCP class from mcp.server.fastmcp
- We need to decorate definition with @mcp.tool() to make it a MCP tool
- Doc string is very important & mandatory in the tools - LLM will identify the required tool based on this doc string only. Hence it is very important.
Note :
We need to test the MCP server before deploying it in production. There are multiple tools to test the MCP server but the recommended tool is MCP inspector.
How to invoke MCP Inspector ?
Command : mcp dev <filename.py> - then it will open below MCP Inspector window.
Lets understand below components of MCP Inspector :
- Transport Type
- STDIO
- Means MCP server created & deployed in local machine
- Can accessible by you only
- SSE
- Legacy version of StreamableHTTP
- Means MCP server created in local machine but deployed in remote machine(cloud)
- Can be accessible by anyone as it is not in local
- People are not using SSE these days
- StreamableHTTP
- Means MCP server created in local machine but deployed in remote machine(cloud)
- Can be accessible by anyone as it is not in local
Once MCP server window is UP, click on "Connect" then you will see that MCP server is connected to MCP Inspector.
You can see Resources, Prompts, Tools listed on the top right of screen (as shown in the above image).
How to test our tools ?
- Click on Tools button, then it will be navigated to Tools window.
- Then we can see our implemented tools on the screen as shown in the below image
- Click on any tools on the bottom and you will see an option to enter the input accepted by your tool
- Give proper "input" and click on "Run Tool" tool output will be displayed on the bottom as shown in the below image
This is how you test your TOOLS in your MCP server using MCP Inspector. DO NOT deploy without testing your MCP tools.
Implementing a MCP server for basic calculations :
# ─────────────────────────────────────────────────────────
# CALCULATOR MCP SERVER
# Fixed: removed all print() to stdout — corrupts stdio pipe
# All logging goes to stderr only
# ─────────────────────────────────────────────────────────
import sys
from mcp.server.fastmcp import FastMCP
def log(msg):
"""Log to stderr only — stdout is reserved for MCP protocol."""
print(msg, file=sys.stderr, flush=True)
mcp = FastMCP("Calculator")
@mcp.tool()
def add(a: float, b: float) -> float:
"""Add two numbers."""
return a + b
@mcp.tool()
def subtract(a: float, b: float) -> float:
"""Subtract b from a."""
return a - b
@mcp.tool()
def multiply(a: float, b: float) -> float:
"""Multiply two numbers."""
return a * b
@mcp.tool()
def divide(a: float, b: float) -> float:
"""Divide a by b. Raises error if b is zero."""
if b == 0:
raise ValueError("Cannot divide by zero.")
return a / b
if __name__ == "__main__":
log("Calculator MCP Server starting...")
mcp.run(transport="stdio")
Implementing a Weather MCP server :
"""
Weather MCP Server using SerpAPI
pip install mcp httpx python-dotenv
"""
import os
import sys
import httpx
from pathlib import Path
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
# ── Load .env from same folder as this file ───────────────
load_dotenv(Path(__file__).parent / ".env")
SERPAPI_KEY = os.getenv("SERPAPI_API_KEY", "")
SERPAPI_URL = "https://serpapi.com/search.json"
# ── CRITICAL: use stderr NOT stdout ───────────────────────
# stdout is used by MCP protocol — any print() to stdout
# corrupts the stdio pipe and causes "Connection closed"
def log(msg):
print(msg, file=sys.stderr, flush=True)
log(f"SERPAPI_KEY : {'SET' if SERPAPI_KEY else 'NOT SET'}")
mcp = FastMCP("Weather Server")
@mcp.tool()
def get_weather(city: str) -> dict:
"""
Get current weather for any city using SerpAPI.
Args:
city: City name e.g. 'Hyderabad', 'London', 'New York'
Returns:
Dict with temperature, humidity, wind, condition and 7-day forecast
"""
log(f"Fetching weather for: {city}")
response = httpx.get(SERPAPI_URL, params={
"q": f"weather in {city}",
"engine": "google",
"api_key": SERPAPI_KEY,
"hl": "en",
})
response.raise_for_status()
data = response.json()
box = data.get("answer_box", {})
if not box or box.get("type") != "weather_result":
return {"error": f"No weather data found for '{city}'"}
forecast = [
{
"day": day.get("day"),
"condition": day.get("weather"),
"high": str(day.get("temperature", {}).get("high", "")) + "°F",
"low": str(day.get("temperature", {}).get("low", "")) + "°F",
"humidity": day.get("humidity"),
"wind": day.get("wind"),
}
for day in box.get("forecast", [])
]
result = {
"location": box.get("location", city),
"date": box.get("date"),
"condition": box.get("weather"),
"temperature": str(box.get("temperature","")) + "°" +
box.get("unit", "Fahrenheit")[0],
"humidity": box.get("humidity"),
"wind": box.get("wind"),
"precipitation": box.get("precipitation"),
"forecast": forecast,
}
log(f"Weather fetched: {result['location']} - {result['temperature']}")
return result
@mcp.tool()
def compare_weather(city1: str, city2: str) -> dict:
"""
Compare current weather between two cities.
Args:
city1: First city e.g. 'Hyderabad'
city2: Second city e.g. 'Mumbai'
Returns:
Side-by-side comparison with warmer city identified
"""
w1 = get_weather(city1)
w2 = get_weather(city2)
return {
city1: w1,
city2: w2,
"warmer": city1 if w1.get("temperature","") > w2.get("temperature","")
else city2,
}
if __name__ == "__main__":
log("Starting Weather MCP Server...")
mcp.run(transport="stdio") # ← explicit stdio transport
Implementing File System Explorer MCP Server :
import os
import sys
from mcp.server.fastmcp import FastMCP
def log(msg):
print(msg, file=sys.stderr, flush=True)
mcp = FastMCP("File Explorer")
@mcp.tool()
def list_files(directory: str = ".") -> list[str]:
"""List all files and folders in a directory."""
if not os.path.isdir(directory):
raise ValueError(f"\'{directory}\' is not a valid directory.")
return os.listdir(directory)
@mcp.tool()
def read_file(filepath: str) -> str:
"""Read and return the contents of a text file."""
if not os.path.isfile(filepath):
raise FileNotFoundError(f"File not found: {filepath}")
with open(filepath, "r", encoding="utf-8") as f:
return f.read()
@mcp.tool()
def write_file(filepath: str, content: str) -> str:
"""Write content to a file (creates or overwrites)."""
with open(filepath, "w", encoding="utf-8") as f:
f.write(content)
return f"Written {len(content)} characters to \'{filepath}\'."
@mcp.tool()
def file_info(filepath: str) -> dict:
"""Return size, extension, and modification time of a file."""
if not os.path.exists(filepath):
raise FileNotFoundError(f"Not found: {filepath}")
stat = os.stat(filepath)
return {
"name": os.path.basename(filepath),
"extension": os.path.splitext(filepath)[1],
"size_bytes": stat.st_size,
"modified": stat.st_mtime,
}
if __name__ == "__main__":
log("File Explorer MCP Server starting...")
mcp.run(transport="stdio")
Comparison between Tools, Resources & Prompts :
Implementation of Ecommerce RAG+MCP server :
# ────────────────────────────────────────────────────────────
# EXAMPLE 6 — ECOMMERCE RAG HELPER (intermediate, realistic)
# Concepts: SQLite for orders, semantic intent detection,
# Resources with URI templates, Prompts,
# structured tool responses, combining tools
# ────────────────────────────────────────────────────────────
import sqlite3
import re
from datetime import datetime, timedelta
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Ecommerce Assistant")
# ── Tiny in-memory SQLite with seed data ──────────────────
def _get_db() -> sqlite3.Connection:
conn = sqlite3.connect(":memory:", check_same_thread=False)
conn.row_factory = sqlite3.Row
conn.executescript("""
CREATE TABLE IF NOT EXISTS orders (
id TEXT PRIMARY KEY,
customer TEXT,
product TEXT,
status TEXT,
amount REAL,
created_at TEXT
);
INSERT OR IGNORE INTO orders VALUES
('ORD001','Alice','Laptop','shipped', 75000, '2025-04-10'),
('ORD002','Alice','Mouse', 'delivered', 1500, '2025-04-01'),
('ORD003','Bob', 'Phone', 'processing',45000, '2025-04-14'),
('ORD004','Bob', 'Case', 'cancelled', 800, '2025-04-05');
""")
return conn
_db = _get_db()
# ── Intent helpers (reuse M6 logic) ───────────────────────
_CANCEL_PAT = re.compile(
r"cancel|abort|stop\s+order|withdraw\s+order|undo\s+order", re.I
)
_STATUS_PAT = re.compile(
r"status|where\s+is|track|shipped|deliver", re.I
)
def _detect_intent(query: str) -> str:
if _CANCEL_PAT.search(query): return "cancel"
if _STATUS_PAT.search(query): return "status"
return "general"
# ── Tools ─────────────────────────────────────────────────
@mcp.tool()
def get_order(order_id: str) -> dict:
"""Fetch full details of a single order by ID."""
row = _db.execute(
"SELECT * FROM orders WHERE id = ?", (order_id.upper(),)
).fetchone()
if not row:
raise ValueError(f"Order '{order_id}' not found.")
return dict(row)
@mcp.tool()
def get_customer_orders(customer_name: str) -> list[dict]:
"""List all orders for a given customer."""
rows = _db.execute(
"SELECT * FROM orders WHERE customer = ?", (customer_name,)
).fetchall()
return [dict(r) for r in rows]
@mcp.tool()
def cancel_order(order_id: str) -> dict:
"""
Cancel an order if eligible.
Orders already 'shipped' or 'delivered' cannot be cancelled.
"""
order = get_order(order_id)
if order["status"] in ("shipped", "delivered"):
return {
"success": False,
"message": (
f"Order {order_id} cannot be cancelled — "
f"it is already {order['status']}. "
"Please initiate a return instead."
),
}
if order["status"] == "cancelled":
return {"success": False, "message": f"Order {order_id} is already cancelled."}
_db.execute(
"UPDATE orders SET status = 'cancelled' WHERE id = ?",
(order_id.upper(),)
)
return {
"success": True,
"order_id": order_id,
"message": f"Order {order_id} cancelled successfully.
Refund of ₹{order['amount']} will be processed in 5–7 days.",
}
@mcp.tool()
def handle_customer_query(customer: str, query: str) -> dict:
"""
Smart dispatcher: detects intent from natural language,
fetches relevant order data, and returns a structured response.
Uses the same intent logic as M6_QueryProcessing.
"""
intent = _detect_intent(query)
orders = get_customer_orders(customer)
if not orders:
return {"intent": intent, "message": f"No orders found for {customer}."}
if intent == "cancel":
# Find the most recent cancellable order
cancellable = [o for o in orders if o["status"] == "processing"]
if cancellable:
result = cancel_order(cancellable[0]["id"])
return {"intent": "cancel", **result}
return {
"intent": "cancel",
"success": False,
"message": "No cancellable orders found.
Orders must be in 'processing' status.",
}
if intent == "status":
return {
"intent": "status",
"orders": [{"id": o["id"], "product": o["product"],
"status": o["status"]} for o in orders],
"message": f"Found {len(orders)} order(s) for {customer}.",
}
return {
"intent": "general",
"orders": orders,
"message": f"{customer} has {len(orders)} order(s) total.",
}
# ── Resources (expose data as readable URIs) ──────────────
@mcp.resource("orders://{customer}/all")
def customer_orders_resource(customer: str) -> str:
"""URI template: orders://Alice/all → formatted order list."""
orders = get_customer_orders(customer)
if not orders:
return f"No orders for {customer}."
lines = [f"Orders for {customer}:"]
for o in orders:
lines.append(f" {o['id']} | {o['product']:<12} | {o['status']:<12}
| ₹{o['amount']}")
return "\n".join(lines)
# ── Prompts (reusable LLM prompt templates) ───────────────
@mcp.prompt()
def order_support_prompt(customer: str, issue: str) -> str:
"""Generate a support prompt pre-filled with customer context."""
orders = get_customer_orders(customer)
order_summary = "\n".join(
f"- {o['id']}: {o['product']} ({o['status']})" for o in orders
)
return (
f"You are a helpful ecommerce support agent.\n\n"
f"Customer: {customer}\n"
f"Their orders:\n{order_summary}\n\n"
f"Customer issue: {issue}\n\n"
f"Resolve this helpfully and concisely."
)
# ────────────────────────────────────────────────────────────
# QUICK DEMO (run directly to see output)
# ────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("\n" + "=" * 55)
print("EXAMPLE 6 — Ecommerce Assistant")
print("=" * 55)
print("\nFetch order ORD001:")
print(get_order("ORD001"))
print("\nAlice's orders:")
for o in get_customer_orders("Alice"):
print(f" {o['id']} | {o['product']} | {o['status']}")
print("\nCancel ORD003 (processing):")
print(cancel_order("ORD003"))
print("\nCancel ORD001 (already shipped):")
print(cancel_order("ORD001"))
print("\nSmart query — 'I want to cancel my order':")
print(handle_customer_query("Bob", "I want to cancel my order"))
print("\nSmart query — 'where is my order?':")
print(handle_customer_query("Alice", "where is my order?"))
print("\nGenerated support prompt:")
print(order_support_prompt("Alice", "My laptop hasn't arrived yet"))
Explanation :
- We have used a lite weight database called sqlite
- We have created a schema and inserted some records into it
- Interacted with the inserted records using MCP inspector to understand how to deal with DB's using in MCP server
- We created 4 tools, one prompt and one resource
- We can dynamic prompts using MCP prompt and submit to LLM
- Resources are meant to read or get some content but we don't use resources in production
Implementing a Jira MCP Server :
"""
JIRA MCP SERVER
Fixed: removed all print() to stdout — corrupts stdio pipe
All logging redirected to stderr only
"""
import os
import sys
import base64
import httpx
from pathlib import Path
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
# ── Load .env from same folder as this file ───────────────
load_dotenv(Path(__file__).parent / ".env")
def log(msg):
"""Log to stderr only — stdout is reserved for MCP protocol."""
print(msg, file=sys.stderr, flush=True)
# ── Config ─────────────────────────────────────────────────
JIRA_URL = os.getenv("JIRA_URL", "")
JIRA_EMAIL = os.getenv("JIRA_EMAIL", "")
JIRA_TOKEN = os.getenv("JIRA_API_TOKEN", "")
log(f"JIRA_URL : {JIRA_URL}")
log(f"JIRA_EMAIL : {JIRA_EMAIL}")
log(f"JIRA_TOKEN : {'SET' if JIRA_TOKEN else 'NOT SET'}")
mcp = FastMCP(name="jira-reader")
def _auth_header() -> dict:
"""Build Basic Auth header from email + token."""
creds = base64.b64encode(f"{JIRA_EMAIL}:{JIRA_TOKEN}".encode()).decode()
return {
"Authorization": f"Basic {creds}",
"Accept": "application/json",
"Content-Type": "application/json",
}
def _extract_description(adf_node) -> str:
"""Recursively extract plain text from Atlassian Document Format."""
if adf_node is None:
return ""
if isinstance(adf_node, str):
return adf_node
if isinstance(adf_node, dict):
if adf_node.get("type") == "text":
return adf_node.get("text", "")
return " ".join(
_extract_description(child)
for child in adf_node.get("content", [])
)
return ""
@mcp.tool()
def get_jira_issue(issue_key: str) -> dict:
"""
Fetch full details of a Jira issue by key.
Args:
issue_key: Jira issue key e.g. 'KAN-1', 'KAN-2'
Returns:
Dict with summary, status, priority, assignee and description
"""
url = f"{JIRA_URL}/rest/api/3/issue/{issue_key}"
log(f"Fetching issue: {url}")
with httpx.Client() as client:
resp = client.get(url, headers=_auth_header())
log(f"Status: {resp.status_code}")
if resp.status_code == 401:
return {"error": "401 Unauthorized - check JIRA_EMAIL and JIRA_API_TOKEN"}
elif resp.status_code == 404:
return {"error": f"404 Not Found - Issue {issue_key} does not exist"}
elif resp.status_code == 403:
return {"error": "403 Forbidden - no permission to view this issue"}
elif not resp.text.strip():
return {"error": f"Empty response - Status: {resp.status_code}"}
resp.raise_for_status()
data = resp.json()
fields = data.get("fields", {})
assignee = fields.get("assignee") or {}
story_points = (
fields.get("story_points")
or fields.get("customfield_10016")
or fields.get("customfield_10028")
)
return {
"key": data["key"],
"summary": fields.get("summary", ""),
"description": _extract_description(fields.get("description")),
"status": fields.get("status", {}).get("name", "Unknown"),
"priority": fields.get("priority", {}).get("name", "None"),
"assignee": assignee.get("displayName", "Unassigned"),
"story_points": story_points,
"issue_type": fields.get("issuetype", {}).get("name", ""),
"labels": fields.get("labels", []),
"url": f"{JIRA_URL}/browse/{data['key']}",
}
@mcp.tool()
def search_jira_issues(jql: str, max_results: int = 10) -> list[dict]:
"""
Search Jira using JQL (Jira Query Language).
Args:
jql: JQL query e.g. 'project=KAN AND status="In Progress"'
max_results: Max results to return (default 10)
Returns:
List of matching issues
"""
url = f"{JIRA_URL}/rest/api/3/search/jql"
params = {
"jql": jql,
"maxResults": max_results,
"fields": "summary,status,assignee,priority,issuetype",
}
log(f"JQL search: {jql}")
with httpx.Client(follow_redirects=False) as client:
resp = client.get(url, headers=_auth_header(), params=params)
log(f"Status: {resp.status_code}")
if resp.status_code == 400:
return [{"error": f"Bad JQL: {resp.text}"}]
elif resp.status_code == 401:
return [{"error": "Unauthorized - check your token"}]
resp.raise_for_status()
data = resp.json()
issues = []
for issue in data.get("issues", []):
f = issue["fields"]
issues.append({
"key": issue["key"],
"summary": f.get("summary", ""),
"status": f.get("status", {}).get("name", ""),
"assignee": (f.get("assignee") or {}).get("displayName", "Unassigned"),
"type": f.get("issuetype", {}).get("name", ""),
"url": f"{JIRA_URL}/browse/{issue['key']}",
})
return issues
@mcp.tool()
def get_issue_comments(issue_key: str) -> list[dict]:
"""
Fetch all comments for a Jira issue.
Args:
issue_key: Jira issue key e.g. 'KAN-1'
Returns:
List of comments with author, body, and creation time
"""
url = f"{JIRA_URL}/rest/api/3/issue/{issue_key}/comment"
log(f"Fetching comments for: {issue_key}")
with httpx.Client(follow_redirects=False) as client:
resp = client.get(url, headers=_auth_header())
log(f"Status: {resp.status_code}")
if resp.status_code == 404:
return [{"error": f"Issue {issue_key} not found"}]
elif resp.status_code == 401:
return [{"error": "Unauthorized - check your token"}]
elif resp.status_code == 403:
return [{"error": "Forbidden - no permission to view comments"}]
resp.raise_for_status()
data = resp.json()
comments = data.get("comments", [])
if not comments:
return [{"message": f"No comments found on {issue_key}"}]
return [
{
"author": c.get("author", {}).get("displayName", "Unknown"),
"body": _extract_description(c.get("body")),
"created": c.get("created", ""),
}
for c in comments
]
if __name__ == "__main__":
log("Jira MCP Server starting...")
mcp.run(transport="stdio")
Explanation :
- Before executing above program, follow below steps
- Create a Jira account using your email ID : Jira | Project Management for the AI Era | Atlassian
- Create an API key for your Jira account : https://id.atlassian.com/manage-profile/security/api-tokens
- Keep your email ID, API key & Jira space link in the .env file
JIRA_URL=<Jira_URL>
JIRA_EMAIL=<Your_Email_ID_Used_To_Create_Jira_Account>
JIRA_API_TOKEN=<Your_Jira_API_Token>
- Once account is created, create few tasks, assign them to yourself and add some comments
- Then try to execute above code.
Implementing GitHub MCP Server :
"""
GITHUB MCP SERVER
Fixed: removed all print() to stdout — corrupts stdio pipe
All logging redirected to stderr only
"""
import os
import sys
import base64
import httpx
from pathlib import Path
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
load_dotenv(Path(__file__).parent / ".env")
def log(msg):
print(msg, file=sys.stderr, flush=True)
GITHUB_TOKEN = os.getenv("GITHUB_API_TOKEN", "")
GITHUB_API = "https://api.github.com"
log(f"GITHUB_TOKEN: {'SET' if GITHUB_TOKEN else 'NOT SET'}")
mcp = FastMCP(name="github-reader")
def _headers() -> dict:
return {
"Authorization": f"Bearer {GITHUB_TOKEN}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
@mcp.tool()
def read_github_file(owner: str, repo: str, file_path: str, branch: str = "main")
-> dict:
"""
Read the content of a file from a GitHub repository.
Args:
owner: GitHub username or org e.g. 'gvbigdata'
repo: Repository name e.g. 'GenAI_AgenticAI_RAG'
file_path: Path to file e.g. 'README.md'
branch: Branch name (default: 'main')
Returns:
Dict with file name, path, size, and content
"""
url = f"{GITHUB_API}/repos/{owner}/{repo}/contents/{file_path}"
log(f"Reading file: {owner}/{repo}/{file_path}")
with httpx.Client() as client:
resp = client.get(url, headers=_headers(), params={"ref": branch})
resp.raise_for_status()
data = resp.json()
raw_content = base64.b64decode(data["content"]).decode("utf-8", errors="replace")
return {
"name": data["name"],
"path": data["path"],
"size_bytes": data["size"],
"branch": branch,
"html_url": data["html_url"],
"content": raw_content,
}
@mcp.tool()
def list_repo_files(owner: str, repo: str, path: str = "", branch: str = "main")
-> list[dict]:
"""
List all files and directories at a given path in a GitHub repo.
Args:
owner: GitHub username or org
repo: Repository name
path: Directory path ('' for root)
branch: Branch name (default: 'main')
Returns:
List of files and directories
"""
url = f"{GITHUB_API}/repos/{owner}/{repo}/contents/{path}"
log(f"Listing files: {owner}/{repo}/{path}")
with httpx.Client() as client:
resp = client.get(url, headers=_headers(), params={"ref": branch})
resp.raise_for_status()
items = resp.json()
return [
{
"name": item["name"],
"type": item["type"],
"size": item.get("size", 0),
"path": item["path"],
"html_url": item["html_url"],
}
for item in items
]
@mcp.tool()
def search_github_code(owner: str, repo: str, keyword: str, branch: str = "main")
-> list[dict]:
"""
Search for a keyword across all files in a GitHub repository.
Args:
owner: GitHub username or org
repo: Repository name
keyword: Search keyword e.g. 'embedding'
branch: Branch name (default: 'main')
Returns:
List of files containing the keyword with matching lines
"""
tree_url = f"{GITHUB_API}/repos/{owner}/{repo}/git/trees/{branch}"
log(f"Searching '{keyword}' in {owner}/{repo}")
with httpx.Client(follow_redirects=True) as client:
resp = client.get(tree_url, headers=_headers(), params={"recursive": "1"})
if resp.status_code == 404:
return [{"error": f"Repo not found: {owner}/{repo}"}]
resp.raise_for_status()
tree = resp.json()
code_extensions = {
".py", ".js", ".ts", ".java", ".go", ".md", ".txt",
".yaml", ".yml", ".json", ".toml", ".sh", ".ipynb"
}
files = [
item for item in tree.get("tree", [])
if item["type"] == "blob"
and any(item["path"].endswith(ext) for ext in code_extensions)
and item.get("size", 0) < 500000
]
matches = []
with httpx.Client(follow_redirects=True) as client:
for file in files[:30]:
file_url = (
f"https://raw.githubusercontent.com"
f"/{owner}/{repo}/{branch}/{file['path']}"
)
try:
r = client.get(file_url, headers=_headers())
if r.status_code != 200:
continue
content = r.text
if keyword.lower() in content.lower():
matched_lines = [
f"Line {i+1}: {line.strip()}"
for i, line in enumerate(content.splitlines())
if keyword.lower() in line.lower()
][:5]
matches.append({
"path": file["path"],
"html_url": f"https://github.com/{owner}/{repo}/blob/
{branch}/{file['path']}",
"matched_lines": matched_lines,
})
except Exception:
continue
if not matches:
return [{"message": f"No files found containing '{keyword}' in
{owner}/{repo}"}]
return matches
@mcp.tool()
def get_repo_info(owner: str, repo: str) -> dict:
"""
Get metadata for a GitHub repository.
Args:
owner: GitHub username or org
repo: Repository name
Returns:
Dict with stars, forks, language, description
"""
url = f"{GITHUB_API}/repos/{owner}/{repo}"
log(f"Getting repo info: {owner}/{repo}")
with httpx.Client() as client:
resp = client.get(url, headers=_headers())
resp.raise_for_status()
data = resp.json()
return {
"full_name": data["full_name"],
"description": data.get("description", ""),
"language": data.get("language", ""),
"stars": data["stargazers_count"],
"forks": data["forks_count"],
"open_issues": data["open_issues_count"],
"html_url": data["html_url"],
"updated_at": data["updated_at"],
}
if __name__ == "__main__":
log("GitHub MCP Server starting...")
mcp.run(transport="stdio")
- Make sure you have a GitHub account and created couple of directories and some code files in it.
- Create a GitHub token from 'Developer Tools' of setting from your GitHub account and keep it in .env file in your IDE
Flow of a user query in MCP :
- User query will be sent to MCP client (MCP client is LLM in this case)
- Based on the intent of user query, MCP client re-route user query to MCP server
- MCP server will analyze which tool is is best based on the user query and send this request to that particular tool
- Tool process the request and send the response back to MCL client
- MCP client will orchestrate the response and send final response to user.
MCP vs RAG :
- Inputs are API's, websites, real time data - then MCP
- Data change very rarely - then RAG
- We have use cases where we use both MCP + RAG
MCP Decorators :
- @mcp.Tool() - To do some task
- @mcp.Resources() - Just to get some context/data
- @mcp.Prompt() - For prompts in MCP
Transport Type :
- STDIO - Local MCP server
- SSE
- Streamable HTTP - Production grade transport in cloud
How to test MCP server :
- Using MCP Instpector
- How to start MCP Inspector - mcp dev <filename.py>
Note :
Kindly note that, DO NOT deploy MCP server in production without properly testing all test cases. Sometimes there will be connection issues due to configurations etc. Please use any MCP testing tool like MCP Inspector and and make sure all test cases are executed successfully.
Next we will talk about below points :
- How to deploy MCP server into remote machine or repository ?
- How to call remote MCP server locally ?
- What are the production problems when you are using MCP server ?
How to deploy MCP server into remote machine or repository ?
In this example, we are using Git as remote repo to deploy MCP server. Hence we need to verify if Git is installed in your local machine.
In my case, I have already installed as shown in above image. Incase if haven't installed then please follow the steps mentioned here : https://git-scm.com/install/windows
Please find below commands to prepare a folder and files in it for committing to GitHub.
- git init (turn a normal folder into git repo by adding .git file )
- git add . (take all changes in this folder and prepare them for commit)
- git status (current situation report in Git)
- git commit -m "Add MCP servers for Agentic AI" (save changes permanently)
- git remote add origin https://github.com/gvbigdata/GenAI_MCP_Servers.git
- git push -u origin main (uploads your local commits into remote)
Implementing Remote MCP server :
"""
cd "D:\GenAI Content\AI code\6_MCP"
git init
git add .
git status # verify .env is NOT listed
git commit -m "Add MCP servers for Agentic AI"
git remote add origin https://github.com/amathe1/AI-code.git
git push -u origin main
Calculator MCP Client — Server loaded from GitHub
pip install openai mcp python-dotenv
Add OPENAI_API_KEY to .env
Run: python 13_Remote_MCPServer.py
"""
import os
import sys
import json
import asyncio
import subprocess
import tempfile
import shutil
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
load_dotenv(Path(__file__).parent / ".env")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
# ── GitHub config ─────────────────────────────────────────
GITHUB_REPO = "https://github.com/amathe1/AI-code.git"
SERVER_FILE = "02_Calculator_MCPServer.py" # file name inside repo
async def main():
question = "What is 25 multiplied by 48?"
print(f"Question : {question}")
# ── Step 1: Clone repo from GitHub ────────────────────
temp_dir = tempfile.mkdtemp(prefix="mcp_")
print(f"Cloning : {GITHUB_REPO}")
subprocess.run(
["git", "clone", "--depth=1", GITHUB_REPO, temp_dir],
capture_output=True,
)
# ── Step 2: Find server file in cloned repo ───────────
# Search recursively in case it's inside a subfolder
matches = list(Path(temp_dir).rglob(SERVER_FILE))
if not matches:
print(f"❌ {SERVER_FILE} not found in repo")
shutil.rmtree(temp_dir, ignore_errors=True)
return
server_file = str(matches[0])
print(f"Found : {server_file}")
# ── Step 3: Start MCP server from cloned file ─────────
params = StdioServerParameters(
command=sys.executable,
args=[server_file],
env={"PATH": os.environ.get("PATH", "")}
)
try:
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# ── Get tools ─────────────────────────────
tools = await session.list_tools()
print(f"Tools : {[t.name for t in tools.tools]}")
openai_tools = [
{
"type": "function",
"function": {
"name": t.name,
"description": t.description,
"parameters": t.inputSchema,
}
}
for t in tools.tools
]
# ── Ask OpenAI ────────────────────────────
client = OpenAI(api_key=OPENAI_API_KEY)
messages = [
{"role": "system", "content":
"Use tools to answer math questions."},
{"role": "user", "content": question},
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=openai_tools,
tool_choice="auto",
)
message = response.choices[0].message
# ── Call MCP tool ─────────────────────────
if message.tool_calls:
tc = message.tool_calls[0]
tool_name = tc.function.name
tool_args = json.loads(tc.function.arguments)
print(f"Calling : {tool_name}({tool_args})")
result = await session.call_tool(tool_name, tool_args)
result_text = result.content[0].text
print(f"Result : {result_text}")
messages += [
{
"role": "assistant", "content": message.content,
"tool_calls": [{"id": tc.id, "type": "function",
"function": {"name": tc.function.name,
"arguments": tc.function.arguments}}]
},
{"role": "tool", "tool_call_id": tc.id, "content": result_text},
]
final = client.chat.completions.create(model="gpt-4o-mini",
messages=messages)
answer = final.choices[0].message.content
else:
answer = message.content
print(f"Answer : {answer}")
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
print("Cleaned up temp folder ✅")
if __name__ == "__main__":
asyncio.run(main())
Output :
Note :
- We have been talking about MCP but currently UTCP(Universal Tool Calling Protocol) is under development and in future it may take over this part.
- https://www.utcp.io/
- Similarly Google A2A(Agent-to-Agent) protocol is also in use but that's for Agent to Agent calls. We will talk about it in multi-agent development related blogs which will release in next 2-3 weeks.
Now, lets talk about very important concepts in MCP i.e. production problems in MCP.
Production issues in MCP :
1) Tool Design Problem
Why it happens ?
- Teams mirror backend API's instead of business intent
- Teams simply replicate traditional system, they won't capture business intent
- No standard for Tool granularity & different teams expose inconsistent tools
- Same functionality, multiple tools with different names
- LLM should take care of decision, latency will increase because of redundant tools
- This is a common problem in real time
Impact ?
- Too many tool calls - latency & cost will increase
- LLM will be in a confused state !
Solution ?
- Clearly define intent(doc string) of the tool while designing it
Example :
- check_loan_eligibility_tool()
- get_loan_summary()
- apply_home_loan() - Correct way
- apply_loan() !!!! - Incorrect way, no specification - what type of loan ?
Best Practices :
- One Tool and one intent
- Avoid chaining multiple tools for simple task
- Keep input & output in a structured format
2) Wrong Tool Selection
Why it happens ?
- LLM relies on your vague description or doc string
- No constraints in the prompts or Guardrails
- Too many similar tools
- Each developer shouldn't create own tools for similar functionality
Impact ?
- Incorrect execution and business rule violation by exposing internal tools to users
Best Practices :
- Enrich tool meta data (can maintain .JSON files for tools meta data)
Ex:
{
name : "apply_loan",
description : "call only after eligibility is confirmed"
}
- Add prompt rules
- "Never call apply_loan tool before checking_eligibility"
3) Invalid Tool Arguments (*)
Why it happens ?
- LLM output is natural language instead of structured data
- Missing fields or wrong data types
Impact ?
- Wrong results at the end
Best Practices :
- Strict schema enforcement
- while writing prompt, mention how you should see the output
- For example, cutomer_id must be string only
- Auto correction layer
- If user enters Fifty Lakhs - it must convert into a structured way 50,00,000
- Tenth September this year - must be converted into 10/09/2026
- Once converted, must apply retry mechanism and ask user to confirm it
- Then ONLY we should lock it & proceed.
- Also we need to handle date format as well
- 10/09/2026 may be Indian way but in US standard they start with month(not date)
- We need to handle such things very carefully - impact could be very high
4) Multi-step workflow failures
Why it happens ?
- LLM has no memory of the steps
- Complex workflows
Impact ?
- Sometimes LLM will miss some constraints because of complexity of prompt definitions
- Skipped steps
- Incorrect decisions
Best Practices :
- Add state tracking for each workflow in each of the workflows
5) High latency
Why it happens ?
- Sequential API/ tool calls
- Redundant Tool usage - developers created too many tools
Impact ?
- Poor user experience
- Customer will drop off
- Incorrect decisions
Best Practices :
- Parallel API/ tool calls
- Implementing caching mechanism
- Reduce redundant tools
6) API Failures
Why it happens ?
- Network Issues
Impact ?
- Broken system
- User frustration
Best Practices :
- Retry with backoff
- retry 3 times - could be 'n' of times
- Fallback strategy
- Graceful response
- if system unavailable - send information to customer "Sorry! Unable to fetch data right now"
7) Security risks
Why it happens ?
- LLM has a direct access to all tools
- No authorization layer
Impact ?
- Financial loss
- Compliance violation
Best Practices :
- RBAC (Role Based access)
- Add approval layer
- Implement API gateway firewall for security - it adds some cost
8) Tool Explosion
Tool developed for one intent, later keep on adding additional functionality to same tool.
Why it happens ?
- Continuous changes to tools
- No appropriate governance
Impact ?
- LLM will confuse
- Reduce the accuracy
Best Practices :
- Tool categorization - make searching easy
- Loan Tools - Home, Car, Personal, Gold loans
- Customer Tools - PII, churn data etc.
- Payment Tools - UPI, Credit card, Net banking etc.
- Tool registry with governance - centralized repository with standard guidelines
- Tool versioning information
- It won't allow redundant tools
- Strictly defining governance policy
9) Debugging difficulty
Why it happens ?
- Multi layer system (LLM + MCP + Tools + API + Prompts)
Impact ?
- Hard to fix production issues
Best Practices :
- Enable full trace logging
- Add observability tools like Langsmith, Opik & Open Telemetry etc.
10) Data Inconsistency
Why it happens ?
- Multiple data sources
- Different update cycles leads to data inconsistency
Impact ?
- Wrong decisions, loss of trust
Best Practices :
- Maintain source of truth
- Data timestamping
11) Weak Prompting
Why it happens ?
- Generic instructions instead of intent based instructions
- Not adding constraints
Impact ?
- Unpredictable behavior
Best Practices :
- Write structured prompt
- step-1
- step-2
- step-3
- Add constraints & Tone
- Few shot prompting/ examples
12) Cost Explosion
Why it happens ?
- Too many tool calls, increasing ambiguity
- Large prompts
Impact ?
- Very high operational cost
Best Practices :
- Limit tool calls
- Apply smart routing
- Use cache
13) Versioning Issues
Why it happens ?
- API evolves continuously
- Open AI released 5.5 last week, 5.4 released 40 days back
Impact ?
- System failures
- Output will vary
Best Practices :
- Maintain same API
- Incase if you are changing any APIs, then clearly communicate deprecation plan to user to set the expectations
All the above problems are diagrammatically represented in the below image :
Conclusion :
We have successfully completed MCP concepts. We will talk about Guardrails in my next blog.
Thank you for reading this blog !
Arun Mathe
Comments
Post a Comment