Establishing an MCP-driven AI Agent using Gemini and MCP Proxy Framework: A Step-by-Step Implementation Guide
In this tutorial, we use MCP Proxy and Gemini. We first set up a powerful environment with all the necessary dependencies, and then implement an MCP tool server that provides structured services such as web search, data analysis, code execution, and weather information. By connecting these tools to a Gemini-powered MCP client, we demonstrate how context-aware reasoning can be used in conjunction with external tool execution. Throughout the process, we highlighted the seamless integration between asynchronous design, tool architecture definition, and the generation capabilities of the MCP layer and Gemini to ensure that our agents remain modular, scalable and production ready. Check The complete code is here.
import subprocess
import sys
import os
from typing import Dict, List, Any, Optional, Union
import json
import asyncio
from datetime import datetime
import logging
def install_packages():
"""Install required packages for the tutorial"""
packages = [
'mcp',
'google-generativeai',
'requests',
'beautifulsoup4',
'matplotlib',
'numpy',
'websockets',
'pydantic'
]
for package in packages:
try:
subprocess.check_call([sys.executable, "-m", "pip", "install", package])
print(f"✅ Successfully installed {package}")
except subprocess.CalledProcessError as e:
print(f"❌ Failed to install {package}: {e}")
install_packages()We first define an install_packages function that specifies all dependencies required by our tutorial, including MCP proxy, Gemini and support libraries. We then run this feature to automatically install each package to make sure our environment is fully prepared before proceeding. Check The complete code is here.
import google.generativeai as genai
import requests
from bs4 import BeautifulSoup
import matplotlib.pyplot as plt
import numpy as np
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import TextContent, ImageContent, EmbeddedResource
import mcp.types as types
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)We import all the core libraries we need, from Gemini and web scraping utilities to visualization and numerical tools. We also introduced the MCP proxy module for protocol communication and configuration logging so that we can track the proxy’s execution flow in real time. Check The complete code is here.
class MCPToolServer:
"""MCP Server that provides tools for the AI agent"""
def __init__(self):
self.tools = {
"web_search": {
"name": "web_search",
"description": "Search the web for information",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query"}
},
"required": ["query"]
}
},
"data_analysis": {
"name": "data_analysis",
"description": "Analyze data and create visualizations",
"inputSchema": {
"type": "object",
"properties": {
"data_type": {"type": "string", "description": "Type of analysis"},
"parameters": {"type": "object", "description": "Analysis parameters"}
},
"required": ["data_type"]
}
},
"code_execution": {
"name": "code_execution",
"description": "Execute or generate code",
"inputSchema": {
"type": "object",
"properties": {
"language": {"type": "string", "description": "Programming language"},
"task": {"type": "string", "description": "Code task description"}
},
"required": ["language", "task"]
}
},
"weather_info": {
"name": "weather_info",
"description": "Get weather information",
"inputSchema": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "Location for weather"}
},
"required": ["location"]
}
}
}
async def list_tools(self) -> List[types.Tool]:
"""Return list of available tools"""
return [types.Tool(**tool) for tool in self.tools.values()]
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> List[types.TextContent]:
"""Execute a tool and return results"""
if name == "web_search":
return await self._web_search(arguments.get("query", ""))
elif name == "data_analysis":
return await self._data_analysis(arguments.get("data_type", ""), arguments.get("parameters", {}))
elif name == "code_execution":
return await self._code_execution(arguments.get("language", ""), arguments.get("task", ""))
elif name == "weather_info":
return await self._weather_info(arguments.get("location", ""))
else:
return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
async def _web_search(self, query: str) -> List[types.TextContent]:
"""Perform web search"""
try:
search_url = f" ', '%20')}"
headers = {'User-Agent': 'Mozilla/5.0 (compatible; MCP Agent)'}
response = requests.get(search_url, headers=headers, timeout=10)
if response.status_code == 200:
soup = BeautifulSoup(response.content, 'html.parser')
paragraphs = soup.find_all('p')[:3]
content = "n".join([p.get_text().strip() for p in paragraphs if p.get_text().strip()])
result = f"🔍 Web search results for '{query}':nn{content[:500]}..."
else:
result = f"❌ Web search failed with status: {response.status_code}"
except Exception as e:
result = f"❌ Web search error: {str(e)}"
return [types.TextContent(type="text", text=result)]
async def _data_analysis(self, data_type: str, parameters: Dict) -> List[types.TextContent]:
"""Perform data analysis"""
try:
if "sine" in data_type.lower() or "wave" in data_type.lower():
x = np.linspace(0, 4*np.pi, 100)
y = np.sin(x) + np.random.normal(0, 0.1, 100)
title = "Sine Wave Analysis"
else:
x = np.random.normal(0, 1, 100)
y = np.random.normal(0, 1, 100)
title = "Random Data Analysis"
plt.figure(figsize=(10, 6))
plt.scatter(x, y, alpha=0.6)
plt.title(f"📊 {title}")
plt.xlabel("X Values")
plt.ylabel("Y Values")
plt.grid(True, alpha=0.3)
plt.show()
stats = {
"mean_x": np.mean(x),
"mean_y": np.mean(y),
"std_x": np.std(x),
"std_y": np.std(y),
"correlation": np.corrcoef(x, y)[0,1]
}
result = f"📊 Data Analysis Results:n"
result += f"Dataset: {title}n"
result += f"Sample size: {len(x)}n"
result += f"X - Mean: {stats['mean_x']:.3f}, Std: {stats['std_x']:.3f}n"
result += f"Y - Mean: {stats['mean_y']:.3f}, Std: {stats['std_y']:.3f}n"
result += f"Correlation: {stats['correlation']:.3f}n"
result += f"n📈 Visualization displayed above!"
except Exception as e:
result = f"❌ Data analysis error: {str(e)}"
return [types.TextContent(type="text", text=result)]
async def _code_execution(self, language: str, task: str) -> List[types.TextContent]:
"""Handle code generation/execution"""
try:
if language.lower() == "python":
if "fibonacci" in task.lower():
code=""'def fibonacci(n):
"""Generate fibonacci sequence up to n terms"""
if n List[types.TextContent]:
"""Get weather information"""
weather_data = {
"temperature": np.random.randint(15, 30),
"condition": np.random.choice(["Sunny", "Cloudy", "Rainy", "Partly Cloudy"]),
"humidity": np.random.randint(40, 80),
"wind_speed": np.random.randint(5, 25)
}
result = f"🌤️ Weather for {location}:n"
result += f"Temperature: {weather_data['temperature']}°Cn"
result += f"Condition: {weather_data['condition']}n"
result += f"Humidity: {weather_data['humidity']}%n"
result += f"Wind Speed: {weather_data['wind_speed']} km/hn"
result += f"n📝 Note: This is simulated data. For real weather, use a weather API service."
return [types.TextContent(type="text", text=result)]We designed the McPtoolServer class, which defines and manages all the tools that an agent can use, including web search, data analysis, code execution, and weather information. We implement an asynchronous approach for each tool, allowing the agent to perform requested operations such as fetching Wikipedia text, generating visualizations, performing Python snippets or simulated weather data, and returning the results in a structured format. This structure makes our MCP server modular and easy to scale to add more tools in the future. Check The complete code is here.
class MCPAgent:
"""AI Agent using MCP (Model Context Protocol)"""
def __init__(self, gemini_api_key: Optional[str] = None):
self.gemini_api_key = gemini_api_key or self._get_api_key()
self.mcp_server = MCPToolServer()
self.conversation_history = []
if self.gemini_api_key:
genai.configure(api_key=self.gemini_api_key)
self.model = genai.GenerativeModel('gemini-1.5-flash')
print("✅ MCP Agent initialized with Gemini!")
else:
self.model = None
print("⚠️ MCP Agent initialized without Gemini (limited functionality)")
def _get_api_key(self) -> Optional[str]:
"""Get Gemini API key"""
api_key = os.environ.get('GEMINI_API_KEY')
if not api_key:
print("📝 Get your free API key from:
api_key = input("Enter your Gemini API key (or press Enter to skip): ").strip()
return api_key if api_key else None
async def process_request(self, user_input: str) -> str:
"""Process user request using MCP tools and Gemini"""
self.conversation_history.append({"role": "user", "content": user_input})
available_tools = await self.mcp_server.list_tools()
tool_descriptions = "n".join([f"- {tool.name}: {tool.description}" for tool in available_tools])
if self.model:
analysis_prompt = f"""
User request: "{user_input}"
Available MCP tools:
{tool_descriptions}
Should I use a tool for this request? If yes, specify:
1. Tool name (exact match)
2. Arguments as JSON
If no tool needed, respond with "NO_TOOL".
Format: TOOL_NAME|{{"argument": "value"}}
"""
analysis = self.model.generate_content(analysis_prompt).text.strip()
if analysis != "NO_TOOL" and "|" in analysis:
try:
tool_name, args_json = analysis.split("|", 1)
tool_name = tool_name.strip()
arguments = json.loads(args_json)
tool_results = await self.mcp_server.call_tool(tool_name, arguments)
tool_output = "n".join([content.text for content in tool_results])
final_prompt = f"""
User asked: "{user_input}"
I used the {tool_name} tool and got this result:
{tool_output}
Please provide a helpful response that incorporates this information.
"""
response = self.model.generate_content(final_prompt).text
except Exception as e:
response = f"❌ Error using MCP tool: {str(e)}nnLet me help you directly instead.n"
response += self.model.generate_content(user_input).text
else:
response = self.model.generate_content(user_input).text
else:
response = f"🤖 MCP Agent received: {user_input}n"
response += "Available tools: " + ", ".join([tool.name for tool in available_tools])
response += "n💡 Configure Gemini API for full functionality!"
self.conversation_history.append({"role": "assistant", "content": response})
return responseWe define a MCPAGENT that connects Gemini to our MCP tool server and maintains conversation history, allowing us to reason, decide on tools, execute and synthesize results. We get the Gemini API key, configure the model, and in Process_request, we prompt Gemini to select the tool (or answer directly), run the selected tool asynchronously, and form the final response based on the tool output in the tool output. Check The complete code is here.
async def run_mcp_demo():
"""Run comprehensive MCP Agent demo"""
print("🚀 MCP Agent Demo Starting!")
print("=" * 50)
agent = MCPAgent()
demo_queries = [
"Search for information about machine learning",
"Create a data visualization with sine wave analysis",
"What's the weather like in New York?",
"Explain how artificial intelligence works"
]
print("n🧪 Running MCP Tool Demonstrations:")
print("-" * 40)
for i, query in enumerate(demo_queries, 1):
print(f"n📝 Query {i}: {query}")
print("-" * 30)
response = await agent.process_request(query)
print(response)
if i We ran a script demo that initialized McPagent, performed a series of representative queries, and printed Gemini-Driven, the tool-initiated response using a brief pause between runs. We then enter an interactive loop where we can list tools, send arbitrary hints and observe end-to-end MCP orchestration before printing a concise review of the concepts covered.
In short, we now have a comprehensive MCP proxy that dynamically determines when to use external tools and how to merge their output into meaningful responses. We validate the proxy in multiple queries, demonstrating its ability to search, analyze, generate and simulate real-world interactions with Gemini as an inference engine. By combining the structured MCP protocol with Gemini’s flexibility, we created a template for building powerful AI systems that are both interactive and technology-based.
Check The complete code is here. Check out ours anytime Tutorials, codes and notebooks for github pages. Also, please stay tuned for us twitter And don’t forget to join us 100K+ ml reddit And subscribe Our newsletter.
Asif Razzaq is CEO of Marktechpost Media Inc. As a visionary entrepreneur and engineer, ASIF is committed to harnessing the potential of artificial intelligence to achieve social benefits. His recent effort is to launch Marktechpost, an artificial intelligence media platform that has an in-depth coverage of machine learning and deep learning news that can sound both technically, both through technical voices and be understood by a wide audience. The platform has over 2 million views per month, demonstrating its popularity among its audience.
