On this tutorial, we stroll via constructing a complicated AI agent utilizing the mcp-agent and Gemini. We begin by establishing a sturdy setting with all the mandatory dependencies after which implement an MCP device server that gives structured providers resembling net search, information evaluation, code execution, and climate info. By wiring these instruments into an MCP consumer powered by Gemini, we reveal how context-aware reasoning could be mixed with exterior device execution. All through, we emphasize asynchronous design, device schema definition, and seamless integration between the MCP layer and Gemini’s generative capabilities, making certain our agent stays modular, extensible, and production-ready. Try the FULL CODES right here.
import subprocess
import sys
import os
from typing import Dict, Listing, Any, Optionally available, Union
import json
import asyncio
from datetime import datetime
import logging
def install_packages():
"""Set up required packages for the tutorial"""
packages = [
'mcp',
'google-generativeai',
'requests',
'beautifulsoup4',
'matplotlib',
'numpy',
'websockets',
'pydantic'
]
for bundle in packages:
attempt:
subprocess.check_call([sys.executable, "-m", "pip", "install", package])
print(f"✅ Efficiently put in {bundle}")
besides subprocess.CalledProcessError as e:
print(f"❌ Failed to put in {bundle}: {e}")
install_packages()We start by defining an install_packages perform that specifies all of the dependencies required for our tutorial, together with mcp-agent, Gemini, and supporting libraries. We then run this perform to robotically set up every bundle, making certain our surroundings is absolutely ready earlier than continuing additional. Try the FULL CODES right 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.consumer.stdio import stdio_client
from mcp.sorts import TextContent, ImageContent, EmbeddedResource
import mcp.sorts as sorts
logging.basicConfig(stage=logging.INFO)
logger = logging.getLogger(__name__)We import all of the core libraries we want, from Gemini and net scraping utilities to visualization and numerical instruments. We additionally deliver within the mcp-agent modules for protocol communication and configure logging in order that we will monitor our agent’s execution circulate in actual time. Try the FULL CODES right here.
class MCPToolServer:
"""MCP Server that gives instruments for the AI agent"""
def __init__(self):
self.instruments = {
"web_search": {
"title": "web_search",
"description": "Search the online for info",
"inputSchema": {
"kind": "object",
"properties": {
"question": {"kind": "string", "description": "Search question"}
},
"required": ["query"]
}
},
"data_analysis": {
"title": "data_analysis",
"description": "Analyze information and create visualizations",
"inputSchema": {
"kind": "object",
"properties": {
"data_type": {"kind": "string", "description": "Sort of research"},
"parameters": {"kind": "object", "description": "Evaluation parameters"}
},
"required": ["data_type"]
}
},
"code_execution": {
"title": "code_execution",
"description": "Execute or generate code",
"inputSchema": {
"kind": "object",
"properties": {
"language": {"kind": "string", "description": "Programming language"},
"process": {"kind": "string", "description": "Code process description"}
},
"required": ["language", "task"]
}
},
"weather_info": {
"title": "weather_info",
"description": "Get climate info",
"inputSchema": {
"kind": "object",
"properties": {
"location": {"kind": "string", "description": "Location for climate"}
},
"required": ["location"]
}
}
}
async def list_tools(self) -> Listing[types.Tool]:
"""Return record of accessible instruments"""
return [types.Tool(**tool) for tool in self.tools.values()]
async def call_tool(self, title: str, arguments: Dict[str, Any]) -> Listing[types.TextContent]:
"""Execute a device and return outcomes"""
if title == "web_search":
return await self._web_search(arguments.get("question", ""))
elif title == "data_analysis":
return await self._data_analysis(arguments.get("data_type", ""), arguments.get("parameters", {}))
elif title == "code_execution":
return await self._code_execution(arguments.get("language", ""), arguments.get("process", ""))
elif title == "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, question: str) -> Listing[types.TextContent]:
"""Carry out net search"""
attempt:
search_url = f"https://www.wikipedia.org/wiki/Particular:Search?search={question.change(' ', '%20')}"
headers = {'Consumer-Agent': 'Mozilla/5.0 (suitable; MCP Agent)'}
response = requests.get(search_url, headers=headers, timeout=10)
if response.status_code == 200:
soup = BeautifulSoup(response.content material, 'html.parser')
paragraphs = soup.find_all('p')[:3]
content material = "n".be a part of([p.get_text().strip() for p in paragraphs if p.get_text().strip()])
outcome = f"🔍 Net search outcomes for '{question}':nn{content material[:500]}..."
else:
outcome = f"❌ Net search failed with standing: {response.status_code}"
besides Exception as e:
outcome = f"❌ Net search error: {str(e)}"
return [types.TextContent(type="text", text=result)]
async def _data_analysis(self, data_type: str, parameters: Dict) -> Listing[types.TextContent]:
"""Carry out information evaluation"""
attempt:
if "sine" in data_type.decrease() or "wave" in data_type.decrease():
x = np.linspace(0, 4*np.pi, 100)
y = np.sin(x) + np.random.regular(0, 0.1, 100)
title = "Sine Wave Evaluation"
else:
x = np.random.regular(0, 1, 100)
y = np.random.regular(0, 1, 100)
title = "Random Knowledge Evaluation"
plt.determine(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.present()
stats = {
"mean_x": np.imply(x),
"mean_y": np.imply(y),
"std_x": np.std(x),
"std_y": np.std(y),
"correlation": np.corrcoef(x, y)[0,1]
}
outcome = f"📊 Knowledge Evaluation Outcomes:n"
outcome += f"Dataset: {title}n"
outcome += f"Pattern dimension: {len(x)}n"
outcome += f"X - Imply: {stats['mean_x']:.3f}, Std: {stats['std_x']:.3f}n"
outcome += f"Y - Imply: {stats['mean_y']:.3f}, Std: {stats['std_y']:.3f}n"
outcome += f"Correlation: {stats['correlation']:.3f}n"
outcome += f"n📈 Visualization displayed above!"
besides Exception as e:
outcome = f"❌ Knowledge evaluation error: {str(e)}"
return [types.TextContent(type="text", text=result)]
async def _code_execution(self, language: str, process: str) -> Listing[types.TextContent]:
"""Deal with code era/execution"""
attempt:
if language.decrease() == "python":
if "fibonacci" in process.decrease():
code=""'def fibonacci(n):
"""Generate fibonacci sequence as much as n phrases"""
if n Listing[types.TextContent]:
"""Get climate info"""
weather_data = {
"temperature": np.random.randint(15, 30),
"situation": np.random.alternative(["Sunny", "Cloudy", "Rainy", "Partly Cloudy"]),
"humidity": np.random.randint(40, 80),
"wind_speed": np.random.randint(5, 25)
}
outcome = f"🌤️ Climate for {location}:n"
outcome += f"Temperature: {weather_data['temperature']}°Cn"
outcome += f"Situation: {weather_data['condition']}n"
outcome += f"Humidity: {weather_data['humidity']}%n"
outcome += f"Wind Pace: {weather_data['wind_speed']} km/hn"
outcome += f"n📝 Word: That is simulated information. For actual climate, use a climate API service."
return [types.TextContent(type="text", text=result)]We design the MCPToolServer class that defines and manages all of the instruments our agent can use, together with net search, information evaluation, code execution, and climate info. We implement async strategies for every device, enabling the agent to carry out the requested operation, resembling fetching Wikipedia textual content, producing visualizations, executing Python snippets, or simulating climate information, and return the ends in a structured format. This construction makes our MCP server modular and simply extensible for including extra instruments sooner or later. Try the FULL CODES right here.
class MCPAgent:
"""AI Agent utilizing MCP (Mannequin Context Protocol)"""
def __init__(self, gemini_api_key: Optionally available[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.mannequin = genai.GenerativeModel('gemini-1.5-flash')
print("✅ MCP Agent initialized with Gemini!")
else:
self.mannequin = None
print("⚠️ MCP Agent initialized with out Gemini (restricted performance)")
def _get_api_key(self) -> Optionally available[str]:
"""Get Gemini API key"""
api_key = os.environ.get('GEMINI_API_KEY')
if not api_key:
print("📝 Get your free API key from: https://makersuite.google.com/app/apikey")
api_key = enter("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:
"""Course of consumer request utilizing MCP instruments and Gemini"""
self.conversation_history.append({"function": "consumer", "content material": user_input})
available_tools = await self.mcp_server.list_tools()
tool_descriptions = "n".be a part of([f"- {tool.name}: {tool.description}" for tool in available_tools])
if self.mannequin:
analysis_prompt = f"""
Consumer request: "{user_input}"
Obtainable MCP instruments:
{tool_descriptions}
Ought to I take advantage of a device for this request? If sure, specify:
1. Software title (actual match)
2. Arguments as JSON
If no device wanted, reply with "NO_TOOL".
Format: TOOL_NAME|{{"argument": "worth"}}
"""
evaluation = self.mannequin.generate_content(analysis_prompt).textual content.strip()
if evaluation != "NO_TOOL" and "|" in evaluation:
attempt:
tool_name, args_json = evaluation.break up("|", 1)
tool_name = tool_name.strip()
arguments = json.masses(args_json)
tool_results = await self.mcp_server.call_tool(tool_name, arguments)
tool_output = "n".be a part of([content.text for content in tool_results])
final_prompt = f"""
Consumer requested: "{user_input}"
I used the {tool_name} device and obtained this outcome:
{tool_output}
Please present a useful response that comes with this info.
"""
response = self.mannequin.generate_content(final_prompt).textual content
besides Exception as e:
response = f"❌ Error utilizing MCP device: {str(e)}nnLet me assist you to straight as a substitute.n"
response += self.mannequin.generate_content(user_input).textual content
else:
response = self.mannequin.generate_content(user_input).textual content
else:
response = f"🤖 MCP Agent obtained: {user_input}n"
response += "Obtainable instruments: " + ", ".be a part of([tool.name for tool in available_tools])
response += "n💡 Configure Gemini API for full performance!"
self.conversation_history.append({"function": "assistant", "content material": response})
return responseWe outline an MCPAgent that wires Gemini to our MCP device server and maintains dialog historical past, enabling us to cause, determine on a device, execute it, and synthesize the outcome. We fetch the Gemini API key, configure the mannequin, and in process_request, we immediate Gemini to decide on a device (or reply straight), run the chosen device asynchronously, and compose a closing response grounded within the device output. Try the FULL CODES right here.
async def run_mcp_demo():
"""Run complete MCP Agent demo"""
print("🚀 MCP Agent Demo Beginning!")
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🧪 Operating MCP Software Demonstrations:")
print("-" * 40)
for i, question in enumerate(demo_queries, 1):
print(f"n📝 Question {i}: {question}")
print("-" * 30)
response = await agent.process_request(question)
print(response)
if i We run a scripted demo that initializes MCPAgent, executes a collection of consultant queries, and prints Gemini-driven, tool-augmented responses with brief pauses between runs. We then drop into an interactive loop the place we will record instruments, ship arbitrary prompts, and observe end-to-end MCP orchestration, earlier than printing a concise recap of the ideas lined.
In conclusion, we now have a complete MCP agent that dynamically decides when to make use of exterior instruments and how you can merge their outputs into significant responses. We validate the agent throughout a number of queries, showcasing its potential to go looking, analyze, generate, and simulate real-world interactions with Gemini because the reasoning engine. By combining structured MCP protocols with the pliability of Gemini, we create a template for constructing highly effective AI methods which are each interactive and technically grounded.
Try the FULL CODES right here. Be at liberty to take a look at our GitHub Web page for Tutorials, Codes and Notebooks. Additionally, be at liberty to comply with us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our Publication.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its reputation amongst audiences.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the most recent breakthroughs, get unique updates, and join with a worldwide community of future-focused thinkers.
Unlock tomorrow’s developments right now: learn extra, subscribe to our publication, and turn out to be a part of the NextTech group at NextTech-news.com
