Building Advanced MCP (Model Context Protocol) Agents with Multi-Agent Coordination, Context Awareness, and Gemini Integration

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(stage=logging.INFO)
logger = logging.getLogger(__name__)


strive:
   import google.generativeai as genai
   GEMINI_AVAILABLE = True
besides ImportError:
   print("  google-generativeai not put in. Run: pip set up google-generativeai")
   GEMINI_AVAILABLE = False

class AgentFunction(Enum):
   COORDINATOR = "coordinator"
   RESEARCHER = "researcher" 
   ANALYZER = "analyzer"
   EXECUTOR = "executor"


@dataclass
class Message:
   function: str
   content material: str
   timestamp: datetime
   metadata: Dict[str, Any] = None


@dataclass
class AgentContext:
   agent_id: str
   function: AgentFunction
   capabilities: List[str]
   reminiscence: List[Message]
   instruments: List[str]

class MCPAgent:
   """Advanced MCP Agent with developed capabilities - Jupyter Compatible"""
  
   def __init__(self, agent_id: str, function: AgentFunction, api_key: str = None):
       self.agent_id = agent_id
       self.function = function
       self.api_key = api_key
       self.reminiscence = []
       self.context = AgentContext(
           agent_id=agent_id,
           function=function,
           capabilities=self._init_capabilities(),
           reminiscence=[],
           instruments=self._init_tools()
       )
      
       self.mannequin = None
       if GEMINI_AVAILABLE and api_key:
           strive:
               genai.configure(api_key=api_key)
               self.mannequin = genai.GenerativeModel('gemini-pro')
               print(f" Agent {agent_id} initialized with Gemini API")
           besides Exception as e:
               print(f"  Gemini configuration failed: {e}")
               print(" Running in demo mode with simulated responses")
       else:
           print(f" Agent {agent_id} working in demo mode")
      
   def _init_capabilities(self) -> List[str]:
       """Initialize role-specific capabilities"""
       capabilities_map = {
           AgentFunction.COORDINATOR: ["task_decomposition", "agent_orchestration", "priority_management"],
           AgentFunction.RESEARCHER: ["data_gathering", "web_search", "information_synthesis"],
           AgentFunction.ANALYZER: ["pattern_recognition", "data_analysis", "insight_generation"],
           AgentFunction.EXECUTOR: ["action_execution", "result_validation", "output_formatting"]
       }
       return capabilities_map.get(self.function, [])
  
   def _init_tools(self) -> List[str]:
       """Initialize out there instruments primarily based on function"""
       tools_map = {
           AgentFunction.COORDINATOR: ["task_splitter", "agent_selector", "progress_tracker"],
           AgentFunction.RESEARCHER: ["search_engine", "data_extractor", "source_validator"],
           AgentFunction.ANALYZER: ["statistical_analyzer", "pattern_detector", "visualization_tool"],
           AgentFunction.EXECUTOR: ["code_executor", "file_handler", "api_caller"]
       }
       return tools_map.get(self.function, [])
  
   def process_message(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]:
       """Process incoming message with context consciousness - Synchronous model"""
      
       msg = Message(
           function="person",
           content material=message,
           timestamp=datetime.now(),
           metadata=context
       )
       self.reminiscence.append(msg)
      
       immediate = self._generate_contextual_prompt(message, context)
      
       strive:
           if self.mannequin:
               response = self._generate_response_gemini(immediate)
           else:
               response = self._generate_demo_response(message)
          
           response_msg = Message(
               function="assistant",
               content material=response,
               timestamp=datetime.now(),
               metadata={"agent_id": self.agent_id, "function": self.function.worth}
           )
           self.reminiscence.append(response_msg)
          
           return {
               "agent_id": self.agent_id,
               "function": self.function.worth,
               "response": response,
               "capabilities_used": self._analyze_capabilities_used(message),
               "next_actions": self._suggest_next_actions(response),
               "timestamp": datetime.now().isoformat()
           }
          
       besides Exception as e:
           logger.error(f"Error processing message: {e}")
           return {"error": str(e)}
  
   def _generate_response_gemini(self, immediate: str) -> str:
       """Generate response utilizing Gemini API - Synchronous"""
       strive:
           response = self.mannequin.generate_content(immediate)
           return response.textual content
       besides Exception as e:
           logger.error(f"Gemini API error: {e}")
           return self._generate_demo_response(immediate)
  
   def _generate_demo_response(self, message: str) -> str:
       """Generate simulated response for demo functions"""
       role_responses = {
           AgentFunction.COORDINATOR: f"As coordinator, I'll break down the duty: '{message[:50]}...' into manageable elements and assign them to specialised brokers.",
           AgentFunction.RESEARCHER: f"I'll analysis details about: '{message[:50]}...' utilizing my knowledge gathering and synthesis capabilities.",
           AgentFunction.ANALYZER: f"Analyzing the patterns and insights from: '{message[:50]}...' to offer data-driven suggestions.",
           AgentFunction.EXECUTOR: f"I'll execute the mandatory actions for: '{message[:50]}...' and validate the outcomes."
       }
      
       base_response = role_responses.get(self.function, f"Processing: {message[:50]}...")
      
       time.sleep(0.5) 
      
       additional_context = {
           AgentFunction.COORDINATOR: " I've recognized 3 key subtasks and will coordinate their execution throughout the agent workforce.",
           AgentFunction.RESEARCHER: " My analysis signifies a number of related sources and present tendencies on this space.",
           AgentFunction.ANALYZER: " The knowledge exhibits fascinating correlations and actionable insights for determination making.",
           AgentFunction.EXECUTOR: " I've accomplished the requested actions and verified the outputs meet high quality requirements."
       }
      
       return base_response + additional_context.get(self.function, "")
  
   def _generate_contextual_prompt(self, message: str, context: Optional[Dict]) -> str:
       """Generate context-aware immediate primarily based on agent function"""
      
       base_prompt = f"""
       You are a sophisticated AI agent with the function: {self.function.worth}
       Your capabilities: {', '.be a part of(self.context.capabilities)}
       Available instruments: {', '.be a part of(self.context.instruments)}
      
       Recent dialog context:
       {self._get_recent_context()}
      
       Current request: {message}
       """
      
       role_instructions = {
           AgentFunction.COORDINATOR: """
           Focus on breaking down complicated duties, coordinating with different brokers,
           and sustaining general undertaking coherence. Consider dependencies and priorities.
           Provide clear process decomposition and agent assignments.
           """,
           AgentFunction.RESEARCHER: """
           Prioritize correct data gathering, supply verification,
           and complete knowledge assortment. Synthesize findings clearly.
           Focus on present tendencies and dependable sources.
           """,
           AgentFunction.ANALYZER: """
           Focus on sample recognition, knowledge interpretation, and perception technology.
           Provide evidence-based conclusions and actionable suggestions.
           Highlight key correlations and implications.
           """,
           AgentFunction.EXECUTOR: """
           Concentrate on sensible implementation, end result validation,
           and clear output supply. Ensure actions are accomplished successfully.
           Focus on high quality and completeness of execution.
           """
       }
      
       return base_prompt + role_instructions.get(self.function, "")
  
   def _get_recent_context(self, restrict: int = 3) -> str:
       """Get latest dialog context"""
       if not self.reminiscence:
           return "No earlier context"
      
       latest = self.reminiscence[-limit:]
       context_str = ""
       for msg in latest:
           context_str += f"{msg.function}: {msg.content material[:100]}...n"
       return context_str
  
   def _analyze_capabilities_used(self, message: str) -> List[str]:
       """Analyze which capabilities have been probably used"""
       used_capabilities = []
       message_lower = message.decrease()
      
       capability_keywords = {
           "task_decomposition": ["break down", "divide", "split", "decompose"],
           "data_gathering": ["research", "find", "collect", "gather"],
           "pattern_recognition": ["analyze", "pattern", "trend", "correlation"],
           "action_execution": ["execute", "run", "implement", "perform"],
           "agent_orchestration": ["coordinate", "manage", "organize", "assign"],
           "information_synthesis": ["synthesize", "combine", "merge", "integrate"]
       }
      
       for functionality, key phrases in capability_keywords.gadgets():
           if functionality in self.context.capabilities:
               if any(key phrase in message_lower for key phrase in key phrases):
                   used_capabilities.append(functionality)
      
       return used_capabilities
  
   def _suggest_next_actions(self, response: str) -> List[str]:
       """Suggest logical subsequent actions primarily based on response"""
       ideas = []
       response_lower = response.decrease()
      
       if "want extra data" in response_lower or "analysis" in response_lower:
           ideas.append("delegate_to_researcher")
       if "analyze" in response_lower or "sample" in response_lower:
           ideas.append("delegate_to_analyzer") 
       if "implement" in response_lower or "execute" in response_lower:
           ideas.append("delegate_to_executor")
       if "coordinate" in response_lower or "handle" in response_lower:
           ideas.append("initiate_multi_agent_collaboration")
       if "subtask" in response_lower or "break down" in response_lower:
           ideas.append("task_decomposition_required")
          
       return ideas if ideas else ["continue_conversation"]

class MCPAgentSwarm:
   """Multi-agent coordination system - Jupyter Compatible"""
  
   def __init__(self, api_key: str = None):
       self.api_key = api_key
       self.brokers = {}
       self.task_history = []
       self.outcomes = {}
      
   def create_agent(self, agent_id: str, function: AgentFunction) -> MCPAgent:
       """Create and register a brand new agent"""
       agent = MCPAgent(agent_id, function, self.api_key)
       self.brokers[agent_id] = agent
       print(f" Created agent: {agent_id} with function: {function.worth}")
       return agent
  
   def coordinate_task(self, process: str) -> Dict[str, Any]:
       """Coordinate complicated process throughout a number of brokers - Synchronous"""
      
       print(f"n Coordinating process: {process}")
       print("=" * 60)
      
       if "coordinator" not in self.brokers:
           self.create_agent("coordinator", AgentFunction.COORDINATOR)
      
       coordinator = self.brokers["coordinator"]
      
       print("n Step 1: Task Decomposition")
       decomposition = coordinator.process_message(
           f"Decompose this complicated process into subtasks and establish which specialised brokers are wanted: {process}"
       )
       print(f"Coordinator: {decomposition['response']}")
      
       self._ensure_required_agents()
      
       print("n Step 2: Agent Collaboration")
       outcomes = {}
       for agent_id, agent in self.brokers.gadgets():
           if agent_id != "coordinator":
               print(f"n {agent_id.higher()} working...")
               end result = agent.process_message(
                   f"Handle your specialised a part of this process: {process}n"
                   f"Coordinator's steering: {decomposition['response'][:200]}..."
               )
               outcomes[agent_id] = end result
               print(f" {agent_id}: {end result['response'][:150]}...")
      
       print("n Step 3: Final Synthesis")
       final_result = coordinator.process_message(
           f"Synthesize these agent outcomes right into a complete closing output for the duty '{process}':n"
           f"Results abstract: {[f'{k}: {v['response'][:100]}...' for okay, v in outcomes.gadgets()]}"
       )
       print(f"Final Result: {final_result['response']}")
      
       task_record = {
           "process": process,
           "timestamp": datetime.now().isoformat(),
           "decomposition": decomposition,
           "agent_results": outcomes,
           "final_synthesis": final_result,
           "agents_involved": record(self.brokers.keys())
       }
       self.task_history.append(task_record)
      
       return task_record
  
   def _ensure_required_agents(self):
       """Ensure all required agent sorts exist"""
       required_roles = [AgentRole.RESEARCHER, AgentRole.ANALYZER, AgentRole.EXECUTOR]
      
       for function in required_roles:
           agent_id = function.worth
           if agent_id not in self.brokers:
               self.create_agent(agent_id, function)
  
   def get_swarm_status(self) -> Dict[str, Any]:
       """Get present standing of the agent swarm"""
       return {
           "total_agents": len(self.brokers),
           "agent_roles": {support: agent.function.worth for support, agent in self.brokers.gadgets()},
           "tasks_completed": len(self.task_history),
           "last_task": self.task_history[-1]["task"] if self.task_history else "None"
       }

def demo_notebook_compatible():
   """Demonstrate superior MCP agent capabilities - Notebook Compatible"""
  
   print(" Starting Advanced MCP Agent Tutorial")
   print(" Jupyter/Colab Compatible Version")
   print("=" * 60)
  
   API_KEY = None  # Set to your precise key
  
   if not API_KEY:
       print(" Running in DEMO MODE (simulated responses)")
       print(" Set API_KEY variable for actual Gemini AI responses")
       print("-" * 60)
  
   swarm = MCPAgentSwarm(API_KEY)
  
   print("n Demo 1: Single Agent Interaction")
   researcher = swarm.create_agent("research_agent", AgentFunction.RESEARCHER)
  
   end result = researcher.process_message(
       "Research the most recent tendencies in AI agent architectures and multi-agent programs"
   )
   print(f"n Researcher Response:")
   print(f" {end result['response']}")
   print(f" Capabilities Used: {end result['capabilities_used']}")
   print(f" Suggested Next Actions: {end result['next_actions']}")
  
   print("nn Demo 2: Multi-Agent Coordination")
  
   complex_task = """
   Analyze the impression of AI brokers on software program improvement productiveness.
   Include analysis on present instruments, efficiency metrics, future predictions,
   and present actionable suggestions for improvement groups.
   """
  
   coordination_result = swarm.coordinate_task(complex_task)
  
   print("nn Demo 3: Swarm Status")
   standing = swarm.get_swarm_status()
   print(f" Total Agents: {standing['total_agents']}")
   print(f" Agent Roles: {standing['agent_roles']}")
   print(f" Tasks Completed: {standing['tasks_completed']}")
  
   print("n Tutorial Completed Successfully!")
   return swarm


def run_demo():
   """Simple perform to run the demo"""
   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 out the demonstration")