Building a Sophisticated MCP Agent System for Jupyter and Google Colab
This guide walks you through creating an advanced Model Context Protocol (MCP) Agent designed to operate seamlessly within Jupyter notebooks or Google Colab environments. Our focus is on practical, real-world applications emphasizing multi-agent collaboration, contextual awareness, memory retention, and adaptive tool utilization. Each agent is crafted with a distinct specialization-whether coordinating, researching, analyzing, or executing-working collectively as a cohesive swarm capable of tackling intricate challenges.
Setting Up the Environment and Dependencies
We begin by importing critical Python modules for data management, logging, and agent architecture. Logging is configured to facilitate effective debugging and monitoring. Additionally, we verify the presence of the Gemini API, enabling smooth integration if available; otherwise, the system defaults to a simulated demo mode to maintain functionality.
import json
import logging
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
from datetime import datetime
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
try:
import google.generativeai as genai
GEMINI_AVAILABLE = True
except ImportError:
print("⚠️ google-generativeai package not found. Install with: pip install google-generativeai")
GEMINI_AVAILABLE = False
Defining Agent Roles and Core Data Structures
To establish a clear framework, we define AgentRole as an enumeration that assigns specific responsibilities to each agent. The Message dataclass captures conversational exchanges along with timestamps and optional metadata, while AgentContext encapsulates an agent’s identity, role, capabilities, memory, and available tools. This structure ensures efficient management of interactions and state.
class AgentRole(Enum):
COORDINATOR = "coordinator"
RESEARCHER = "researcher"
ANALYZER = "analyzer"
EXECUTOR = "executor"
@dataclass
class Message:
role: str
content: str
timestamp: datetime
metadata: Optional[Dict[str, Any]] = None
@dataclass
class AgentContext:
agent_id: str
role: AgentRole
capabilities: List[str]
memory: List[Message]
tools: List[str]
Implementing the MCPAgent: Role-Aware and Context-Sensitive
The MCPAgent class embodies a role-specific agent optimized for notebook environments. Upon initialization, it assigns capabilities and tools tailored to its role, maintains a memory log of messages, and generates responses informed by context. When the Gemini API is accessible and configured, it leverages real AI-generated content; otherwise, it simulates responses to demonstrate functionality.
class MCPAgent:
"""Role-specialized MCP Agent compatible with Jupyter and Colab"""
def __init__(self, agent_id: str, role: AgentRole, api_key: Optional[str] = None):
self.agent_id = agent_id
self.role = role
self.api_key = api_key
self.memory: List[Message] = []
self.context = AgentContext(
agent_id=agent_id,
role=role,
capabilities=self._initialize_capabilities(),
memory=[],
tools=self._initialize_tools()
)
self.model = None
if GEMINI_AVAILABLE and api_key:
try:
genai.configure(api_key=api_key)
self.model = genai.GenerativeModel('gemini-pro')
print(f"✅ Agent {agent_id} initialized with Gemini API")
except Exception as e:
print(f"⚠️ Gemini API setup failed: {e}")
print("💡 Running in demo mode with simulated responses")
else:
print(f"🎭 Agent {agent_id} operating in demo mode")
def _initialize_capabilities(self) -> List[str]:
role_capabilities = {
AgentRole.COORDINATOR: ["task_decomposition", "agent_orchestration", "priority_management"],
AgentRole.RESEARCHER: ["data_collection", "web_scraping", "information_synthesis"],
AgentRole.ANALYZER: ["pattern_detection", "data_interpretation", "insight_generation"],
AgentRole.EXECUTOR: ["task_execution", "result_verification", "output_formatting"]
}
return role_capabilities.get(self.role, [])
def _initialize_tools(self) -> List[str]:
role_tools = {
AgentRole.COORDINATOR: ["task_splitter", "agent_selector", "progress_monitor"],
AgentRole.RESEARCHER: ["search_engine", "data_extractor", "source_validator"],
AgentRole.ANALYZER: ["statistical_tool", "pattern_recognizer", "visualization_module"],
AgentRole.EXECUTOR: ["code_runner", "file_manager", "api_integrator"]
}
return role_tools.get(self.role, [])
def process_message(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]:
"""Handle incoming messages with contextual awareness (synchronous)"""
user_msg = Message(role="user", content=message, timestamp=datetime.now(), metadata=context)
self.memory.append(user_msg)
prompt = self._build_contextual_prompt(message, context)
try:
if self.model:
response_text = self._generate_response_with_gemini(prompt)
else:
response_text = self._simulate_response(message)
assistant_msg = Message(
role="assistant",
content=response_text,
timestamp=datetime.now(),
metadata={"agent_id": self.agent_id, "role": self.role.value}
)
self.memory.append(assistant_msg)
return {
"agent_id": self.agent_id,
"role": self.role.value,
"response": response_text,
"capabilities_used": self._identify_capabilities(message),
"next_actions": self._recommend_next_steps(response_text),
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error during message processing: {e}")
return {"error": str(e)}
def _generate_response_with_gemini(self, prompt: str) -> str:
try:
response = self.model.generate_content(prompt)
return response.text
except Exception as e:
logger.error(f"Gemini API error: {e}")
return self._simulate_response(prompt)
def _simulate_response(self, message: str) -> str:
role_based_replies = {
AgentRole.COORDINATOR: f"As the coordinator, I will segment the task: '{message[:50]}...' into actionable parts and assign them accordingly.",
AgentRole.RESEARCHER: f"I am gathering data on: '{message[:50]}...' leveraging my research and synthesis skills.",
AgentRole.ANALYZER: f"Examining patterns and insights from: '{message[:50]}...' to deliver data-driven advice.",
AgentRole.EXECUTOR: f"Executing the required steps for: '{message[:50]}...' and ensuring quality validation."
}
base_reply = role_based_replies.get(self.role, f"Processing request: {message[:50]}...")
time.sleep(0.5) # Simulate processing delay
extended_context = {
AgentRole.COORDINATOR: " Identified 3 key subtasks and coordinating their execution among agents.",
AgentRole.RESEARCHER: " Found multiple relevant sources and current trends.",
AgentRole.ANALYZER: " Data reveals significant correlations and actionable insights.",
AgentRole.EXECUTOR: " Actions completed and outputs verified for accuracy."
}
return base_reply + extended_context.get(self.role, "")
def _build_contextual_prompt(self, message: str, context: Optional[Dict]) -> str:
base_prompt = f"""
You are an advanced AI agent with the role: {self.role.value}
Capabilities: {', '.join(self.context.capabilities)}
Tools available: {', '.join(self.context.tools)}
Recent conversation context:
{self._get_recent_memory()}
Current request: {message}
"""
role_guidance = {
AgentRole.COORDINATOR: """
Focus on decomposing complex tasks, orchestrating agent collaboration,
and managing priorities. Provide clear task breakdowns and assignments.
""",
AgentRole.RESEARCHER: """
Emphasize accurate data collection, source validation,
and comprehensive synthesis. Highlight current trends and reliable information.
""",
AgentRole.ANALYZER: """
Concentrate on identifying patterns, interpreting data,
and generating actionable insights. Support conclusions with evidence.
""",
AgentRole.EXECUTOR: """
Prioritize effective implementation, result verification,
and clear output formatting. Ensure thorough completion of tasks.
"""
}
return base_prompt + role_guidance.get(self.role, "")
def _get_recent_memory(self, limit: int = 3) -> str:
if not self.memory:
return "No prior context available."
recent_msgs = self.memory[-limit:]
return "n".join(f"{msg.role}: {msg.content[:100]}..." for msg in recent_msgs)
def _identify_capabilities(self, message: str) -> List[str]:
detected = []
msg_lower = message.lower()
keywords_map = {
"task_decomposition": ["break down", "divide", "split", "decompose"],
"data_collection": ["research", "find", "collect", "gather"],
"pattern_detection": ["analyze", "pattern", "trend", "correlation"],
"task_execution": ["execute", "run", "implement", "perform"],
"agent_orchestration": ["coordinate", "manage", "organize", "assign"],
"information_synthesis": ["synthesize", "combine", "merge", "integrate"]
}
for capability, keywords in keywords_map.items():
if capability in self.context.capabilities and any(k in msg_lower for k in keywords):
detected.append(capability)
return detected
def _recommend_next_steps(self, response: str) -> List[str]:
suggestions = []
resp_lower = response.lower()
if "need more information" in resp_lower or "research" in resp_lower:
suggestions.append("delegate_to_researcher")
if "analyze" in resp_lower or "pattern" in resp_lower:
suggestions.append("delegate_to_analyzer")
if "implement" in resp_lower or "execute" in resp_lower:
suggestions.append("delegate_to_executor")
if "coordinate" in resp_lower or "manage" in resp_lower:
suggestions.append("initiate_multi_agent_collaboration")
if "subtask" in resp_lower or "break down" in resp_lower:
suggestions.append("task_decomposition_required")
return suggestions or ["continue_conversation"]
Coordinating Multiple Agents with MCPAgentSwarm
The MCPAgentSwarm class orchestrates a collection of specialized agents, dynamically creating them as needed and managing complex workflows through task decomposition, inter-agent collaboration, and final result synthesis. It maintains a history of tasks and outcomes, ensures all necessary agent roles are present, and provides a snapshot of the swarm’s current status.
class MCPAgentSwarm:
"""System for managing and coordinating multiple MCP agents"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key
self.agents: Dict[str, MCPAgent] = {}
self.task_history: List[Dict[str, Any]] = []
self.results: Dict[str, Any] = {}
def create_agent(self, agent_id: str, role: AgentRole) -> MCPAgent:
agent = MCPAgent(agent_id, role, self.api_key)
self.agents[agent_id] = agent
print(f"🤖 Created agent '{agent_id}' with role '{role.value}'")
return agent
def coordinate_task(self, task: str) -> Dict[str, Any]:
print(f"n🎯 Coordinating task: {task}")
print("=" * 60)
if "coordinator" not in self.agents:
self.create_agent("coordinator", AgentRole.COORDINATOR)
coordinator = self.agents["coordinator"]
print("n📝 Step 1: Task Decomposition")
decomposition = coordinator.process_message(
f"Decompose this complex task into subtasks and assign appropriate agents: {task}"
)
print(f"Coordinator: {decomposition['response']}")
self._ensure_agents_exist()
print("n🔄 Step 2: Agent Collaboration")
results = {}
for agent_id, agent in self.agents.items():
if agent_id != "coordinator":
print(f"n🤖 {agent_id.upper()} processing...")
result = agent.process_message(
f"Handle your specialized portion of the task: {task}n"
f"Coordinator's instructions: {decomposition['response'][:200]}..."
)
results[agent_id] = result
print(f"✅ {agent_id}: {result['response'][:150]}...")
print("n🎯 Step 3: Final Synthesis")
final_result = coordinator.process_message(
f"Combine these agent outputs into a comprehensive final report for task '{task}':n"
f"Summary: {[f'{k}: {v['response'][:100]}...' for k, v in results.items()]}"
)
print(f"Final Result: {final_result['response']}")
task_record = {
"task": task,
"timestamp": datetime.now().isoformat(),
"decomposition": decomposition,
"agent_results": results,
"final_synthesis": final_result,
"agents_involved": list(self.agents.keys())
}
self.task_history.append(task_record)
return task_record
def _ensure_agents_exist(self):
required_roles = [AgentRole.RESEARCHER, AgentRole.ANALYZER, AgentRole.EXECUTOR]
for role in required_roles:
agent_id = role.value
if agent_id not in self.agents:
self.create_agent(agent_id, role)
def get_swarm_status(self) -> Dict[str, Any]:
return {
"total_agents": len(self.agents),
"agent_roles": {aid: agent.role.value for aid, agent in self.agents.items()},
"tasks_completed": len(self.task_history),
"last_task": self.task_history[-1]["task"] if self.task_history else "None"
}
Demonstration: Showcasing MCP Agent Capabilities in a Notebook
This demonstration script highlights the MCP agent system’s functionality within notebook environments. It begins with a single researcher agent performing a focused query, then advances to a multi-agent collaboration on a complex analysis task, and concludes by displaying the swarm’s current status. The demo gracefully handles the absence of a Gemini API key by simulating responses.
def demo_notebook_compatible():
"""Showcase MCP agent system features in Jupyter/Colab"""
print("🚀 Starting Advanced MCP Agent Demo")
print("📓 Compatible with Jupyter and Google Colab")
print("=" * 60)
API_KEY = None # Replace with your Gemini API key if available
if not API_KEY:
print("💡 Running in DEMO MODE with simulated responses")
print("🔐 Set API_KEY variable to enable real Gemini AI responses")
print("-" * 60)
swarm = MCPAgentSwarm(API_KEY)
print("n📋 Demo 1: Single Agent Interaction")
researcher = swarm.create_agent("research_agent", AgentRole.RESEARCHER)
result = researcher.process_message(
"Research the latest advancements in AI agent architectures and multi-agent frameworks"
)
print("n🔍 Researcher Response:")
print(f"📄 {result['response']}")
print(f"🛠 Capabilities Used: {result['capabilities_used']}")
print(f"➡️ Suggested Next Steps: {result['next_actions']}")
print("n🤝 Demo 2: Multi-Agent Coordination")
complex_task = """
Evaluate the influence of AI agents on software development efficiency.
Include research on current tools, performance metrics, future trends,
and actionable recommendations for engineering teams.
"""
coordination_result = swarm.coordinate_task(complex_task)
print("n📊 Demo 3: Swarm Status")
status = swarm.get_swarm_status()
print(f"🤖 Total Agents: {status['total_agents']}")
print(f"🎭 Agent Roles: {status['agent_roles']}")
print(f"✅ Tasks Completed: {status['tasks_completed']}")
print("n🎉 Demo Completed Successfully!")
return swarm
def run_demo():
"""Execute the MCP agent demonstration"""
return demo_notebook_compatible()
if __name__ == "__main__":
print("🚀 Running MCP Agent Demo...")
swarm = run_demo()
else:
print("📚 MCP Agent Tutorial loaded!")
print("💡 Run: swarm = run_demo() to start the demonstration")
Summary
In this tutorial, we have developed a robust MCP agent framework capable of decomposing complex tasks, coordinating multiple specialized agents, and synthesizing their outputs into actionable insights-all within a notebook-friendly synchronous environment. The system’s memory feature preserves conversational context, while role-based specialization enhances efficiency and adaptability. Integration with the Gemini API offers real AI-powered responses, with a fallback demo mode ensuring uninterrupted operation. This foundation paves the way for sophisticated multi-agent AI applications tailored to dynamic, real-world scenarios.
