Model Context Protocol
Model Context Protocol (MCP)
https://modelcontextprotocol.io/
Below are notes from: https://learn.deeplearning.ai/courses/mcp-build-rich-context-ai-apps-with-anthropic
What is MCP?
MCP = Standardizes how LLM apps interact with external systems.   Stateless connection is useful for cases where we spin up the MCP server on-the-fly (server less compute). If the server is not deployed as ephemeral server then it can use stateful connections as well. 
Example App: Chatbot using Tools (non-MCP)
The tools are defined locally, model selects them and then we execute them locally. We will use MCP tools to change this implementation later. MCP server will do the tool execution.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
import arxiv
import json
import os
from typing import List
from dotenv import load_dotenv
import anthropic
PAPER_DIR = "papers”
def search_papers(topic: str, max_results: int = 5) -> List[str]:
"""
Search for papers on arXiv based on a topic and store their information.
Args:
topic: The topic to search for
max_results: Maximum number of results to retrieve (default: 5)
Returns:
List of paper IDs found in the search
"""
# Use arxiv to find the papers
client = arxiv.Client()
# Search for the most relevant articles matching the queried topic
search = arxiv.Search(
query = topic,
max_results = max_results,
sort_by = arxiv.SortCriterion.Relevance
)
papers = client.results(search)
# Create directory for this topic
path = os.path.join(PAPER_DIR, topic.lower().replace(" ", "_"))
os.makedirs(path, exist_ok=True)
file_path = os.path.join(path, "papers_info.json")
# Try to load existing papers info
try:
with open(file_path, "r") as json_file:
papers_info = json.load(json_file)
except (FileNotFoundError, json.JSONDecodeError):
papers_info = {}
# Process each paper and add to papers_info
paper_ids = []
for paper in papers:
paper_ids.append(paper.get_short_id())
paper_info = {
'title': paper.title,
'authors': [author.name for author in paper.authors],
'summary': paper.summary,
'pdf_url': paper.pdf_url,
'published': str(paper.published.date())
}
papers_info[paper.get_short_id()] = paper_info
# Save updated papers_info to json file
with open(file_path, "w") as json_file:
json.dump(papers_info, json_file, indent=2)
print(f"Results are saved in: {file_path}")
return paper_ids
def extract_info(paper_id: str) -> str:
"""
Search for information about a specific paper across all topic directories.
Args:
paper_id: The ID of the paper to look for
Returns:
JSON string with paper information if found, error message if not found
"""
for item in os.listdir(PAPER_DIR):
item_path = os.path.join(PAPER_DIR, item)
if os.path.isdir(item_path):
file_path = os.path.join(item_path, "papers_info.json")
if os.path.isfile(file_path):
try:
with open(file_path, "r") as json_file:
papers_info = json.load(json_file)
if paper_id in papers_info:
return json.dumps(papers_info[paper_id], indent=2)
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"Error reading {file_path}: {str(e)}")
continue
return f"There's no saved information related to paper {paper_id}."
tools = [
{
"name": "search_papers",
"description": "Search for papers on arXiv based on a topic and store their information.",
"input_schema": {
"type": "object",
"properties": {
"topic": {
"type": "string",
"description": "The topic to search for"
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to retrieve",
"default": 5
}
},
"required": ["topic"]
}
},
{
"name": "extract_info",
"description": "Search for information about a specific paper across all topic directories.",
"input_schema": {
"type": "object",
"properties": {
"paper_id": {
"type": "string",
"description": "The ID of the paper to look for"
}
},
"required": ["paper_id"]
}
}
]
mapping_tool_function = {
"search_papers": search_papers,
"extract_info": extract_info
}
def execute_tool(tool_name, tool_args):
result = mapping_tool_function[tool_name](**tool_args)
if result is None:
result = "The operation completed but didn't return any results."
elif isinstance(result, list):
result = ', '.join(result)
elif isinstance(result, dict):
# Convert dictionaries to formatted JSON strings
result = json.dumps(result, indent=2)
else:
# For any other type, convert using str()
result = str(result)
return result
load_dotenv()
client = anthropic.Anthropic()
def process_query(query):
messages = [{'role': 'user', 'content': query}]
response = client.messages.create(max_tokens = 2024,
model = 'claude-3-7-sonnet-20250219',
tools = tools,
messages = messages)
process_query = True
while process_query:
assistant_content = []
for content in response.content:
if content.type == 'text':
print(content.text)
assistant_content.append(content)
if len(response.content) == 1:
process_query = False
elif content.type == 'tool_use':
assistant_content.append(content)
messages.append({'role': 'assistant', 'content': assistant_content})
tool_id = content.id
tool_args = content.input
tool_name = content.name
print(f"Calling tool {tool_name} with args {tool_args}")
result = execute_tool(tool_name, tool_args)
messages.append({"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_id,
"content": result
}
]
})
response = client.messages.create(max_tokens = 2024,
model = 'claude-3-7-sonnet-20250219',
tools = tools,
messages = messages)
if len(response.content) == 1 and response.content[0].type == "text":
print(response.content[0].text)
process_query = False
def chat_loop():
print("Type your queries or 'quit' to exit.")
while True:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
process_query(query)
print("\n")
except Exception as e:
print(f"\nError: {str(e)}")
— Output:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
Type your queries or 'quit' to exit.
Search for 2 papers on "LLM interpretability"
I'll search for 2 papers on LLM interpretability for you.
Calling tool search_papers with args {'topic': 'LLM interpretability', 'max_results': 2}
Results are saved in: papers/llm_interpretability/papers_info.json
I've found 2 papers related to LLM interpretability. Let me get more information about them for you:
Calling tool extract_info with args {'paper_id': '2412.07992v4'}
Calling tool extract_info with args {'paper_id': '2402.01761v1'}
Here are the details of the 2 papers on LLM interpretability:
### Paper 1: "Concept Bottleneck Large Language Models"
- **Authors**: Chung-En Sun, Tuomas Oikarinen, Berk Ustun, Tsui-Wei Weng
- **Published**: December 11, 2024
- **Summary**: This paper introduces Concept Bottleneck Large Language Models (CB-LLMs), which integrate intrinsic interpretability directly into LLMs. Unlike traditional black-box LLMs that rely on limited post-hoc interpretations, CB-LLMs allow for accurate explanations with scalability and transparency. The authors demonstrate applications in both text classification and text generation, where interpretable neurons enable precise concept detection, controlled generation, and safer outputs. The approach enhances safety, reliability, and trustworthiness of LLMs by allowing users to identify harmful content, steer model behavior, and unlearn undesired concepts.
- **PDF**: http://arxiv.org/pdf/2412.07992v4
### Paper 2: "Rethinking Interpretability in the Era of Large Language Models"
- **Authors**: Chandan Singh, Jeevana Priya Inala, Michel Galley, Rich Caruana, Jianfeng Gao
- **Published**: January 30, 2024
- **Summary**: This position paper reviews existing methods for LLM interpretation (both interpreting LLMs and using LLMs for explanation). The authors argue that despite limitations, LLMs can redefine interpretability with a more ambitious scope across many applications, including auditing the LLMs themselves. They highlight two emerging research priorities: using LLMs to directly analyze new datasets and to generate interactive explanations. The paper also discusses challenges like hallucinated explanations and high computational costs.
- **PDF**: http://arxiv.org/pdf/2402.01761v1
MCP Server
This server needs to handle two main requests from the client:
There are two ways for creating an MCP server:
- low-level implementation: in this approach, you directly define and handle the various types of requests (ListToolsRequest and CallToolRequest). This approach allows you to customize every aspect of your server.
- high-level implementation using FastMCP: FastMCP is a high-level interface that makes building MCP servers faster and simpler. In this approach, you just focus on defining the tools as functions, andFastMCP handles all the protocol details.
https://github.com/jlowin/fastmcp
mcp_project/research_server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import arxiv
import json
import os
from typing import List
from mcp.server.fastmcp import FastMCP
PAPER_DIR = "papers"
# Initialize FastMCP server
mcp = FastMCP("research")
@mcp.tool()
def search_papers(topic: str, max_results: int = 5) -> List[str]:
"""
Search for papers on arXiv based on a topic and store their information.
Args:
topic: The topic to search for
max_results: Maximum number of results to retrieve (default: 5)
Returns:
List of paper IDs found in the search
"""
# Use arxiv to find the papers
client = arxiv.Client()
# Search for the most relevant articles matching the queried topic
search = arxiv.Search(
query = topic,
max_results = max_results,
sort_by = arxiv.SortCriterion.Relevance
)
papers = client.results(search)
# Create directory for this topic
path = os.path.join(PAPER_DIR, topic.lower().replace(" ", "_"))
os.makedirs(path, exist_ok=True)
file_path = os.path.join(path, "papers_info.json")
# Try to load existing papers info
try:
with open(file_path, "r") as json_file:
papers_info = json.load(json_file)
except (FileNotFoundError, json.JSONDecodeError):
papers_info = {}
# Process each paper and add to papers_info
paper_ids = []
for paper in papers:
paper_ids.append(paper.get_short_id())
paper_info = {
'title': paper.title,
'authors': [author.name for author in paper.authors],
'summary': paper.summary,
'pdf_url': paper.pdf_url,
'published': str(paper.published.date())
}
papers_info[paper.get_short_id()] = paper_info
# Save updated papers_info to json file
with open(file_path, "w") as json_file:
json.dump(papers_info, json_file, indent=2)
print(f"Results are saved in: {file_path}")
return paper_ids
@mcp.tool()
def extract_info(paper_id: str) -> str:
"""
Search for information about a specific paper across all topic directories.
Args:
paper_id: The ID of the paper to look for
Returns:
JSON string with paper information if found, error message if not found
"""
for item in os.listdir(PAPER_DIR):
item_path = os.path.join(PAPER_DIR, item)
if os.path.isdir(item_path):
file_path = os.path.join(item_path, "papers_info.json")
if os.path.isfile(file_path):
try:
with open(file_path, "r") as json_file:
papers_info = json.load(json_file)
if paper_id in papers_info:
return json.dumps(papers_info[paper_id], indent=2)
except (FileNotFoundError, json.JSONDecodeError) as e:
print(f"Error reading {file_path}: {str(e)}")
continue
return f"There's no saved information related to paper {paper_id}."
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')
To run the mcp server with inspector:
1
2
3
4
cd mcp_project
uv init
uv add mcp arxiv
npx @modelcontextprotocol/inspector uv run research_server.py
MCP Client
To build a client that connects to the MCP server and interacts with it, you can refer to the following resources:
- https://modelcontextprotocol.io/docs/develop/build-client
- https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-chatbot/mcp_simple_chatbot/main.py
- https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-chatbot/mcp_simple_chatbot/main.py
We will take the functions process_query and chat_loop and wrap them in a MCP_ChatBot class. To enable the chatbot to communicate to the server, you will add a method that connects to the server through an MCP client - connect_to_server_and_run().
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from dotenv import load_dotenv
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
from typing import List
import asyncio
import nest_asyncio
nest_asyncio.apply()
load_dotenv()
class MCP_ChatBot:
def __init__(self):
# Initialize session and client objects
self.session: ClientSession = None
self.anthropic = Anthropic()
self.available_tools: List[dict] = []
async def process_query(self, query):
messages = [{'role':'user', 'content':query}]
response = self.anthropic.messages.create(max_tokens = 2024,
model = 'claude-3-7-sonnet-20250219',
tools = self.available_tools, # tools exposed to the LLM
messages = messages)
process_query = True
while process_query:
assistant_content = []
for content in response.content:
if content.type =='text':
print(content.text)
assistant_content.append(content)
if(len(response.content) == 1):
process_query= False
elif content.type == 'tool_use':
assistant_content.append(content)
messages.append({'role':'assistant', 'content':assistant_content})
tool_id = content.id
tool_args = content.input
tool_name = content.name
print(f"Calling tool {tool_name} with args {tool_args}")
# Call a tool
#result = execute_tool(tool_name, tool_args): not anymore needed
# tool invocation through the client session
result = await self.session.call_tool(tool_name, arguments=tool_args)
messages.append({"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id":tool_id,
"content": result.content
}
]
})
response = self.anthropic.messages.create(max_tokens = 2024,
model = 'claude-3-7-sonnet-20250219',
tools = self.available_tools,
messages = messages)
if(len(response.content) == 1 and response.content[0].type == "text"):
print(response.content[0].text)
process_query= False
async def chat_loop(self):
"""Run an interactive chat loop"""
print("\nMCP Chatbot Started!")
print("Type your queries or 'quit' to exit.")
while True:
try:
query = input("\nQuery: ").strip()
if query.lower() == 'quit':
break
await self.process_query(query)
print("\n")
except Exception as e:
print(f"\nError: {str(e)}")
async def connect_to_server_and_run(self):
# Create server parameters for stdio connection
server_params = StdioServerParameters(
command="uv", # Executable
args=["run", "research_server.py"], # Optional command line arguments
env=None, # Optional environment variables
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
self.session = session
# Initialize the connection
await session.initialize()
# List available tools
response = await session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
self.available_tools = [{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in response.tools]
await self.chat_loop()
async def main():
chatbot = MCP_ChatBot()
await chatbot.connect_to_server_and_run()
if __name__ == "__main__":
asyncio.run(main())
This post is licensed under CC BY 4.0 by the author.