Class MCPAgent
"""Advanced MCP Agent with evolved capabilities - Jupyter Compatible"""
def __init__(self, agent_id: str, role: AgentRole, api_key: str = None):
self.agent_id = agent_id
Self.role = self
self.api_key = api_key
self.memory = []
self.context = AgentContext(
agent_id=agent_id,
role=role,
capabilities=self._init_capabilities(),
memory=[],
tools=self._init_tools()
)
self.model = None
If GEMINI_AVAILABLE AND api_key is:
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 configuration failed: {e}")
print("💡 Running in demo mode with simulated responses")
else:
print(f"🎭 Agent {agent_id} running in demo mode")
def _init_capabilities(self) -> List[str]:
"""Initialize role-specific capabilities"""
capabilities_map = {
AgentRole.COORDINATOR: ["task_decomposition", "agent_orchestration", "priority_management"],
AgentRole.RESEARCHER: ["data_gathering", "web_search", "information_synthesis"],
AgentRole.ANALYZER: ["pattern_recognition", "data_analysis", "insight_generation"],
AgentRole.EXECUTOR: ["action_execution", "result_validation", "output_formatting"]
}
return capabilities_map.get(self.role, [])
def _init_tools(self) -> List[str]:
"""Initialize available tools based on role"""
tools_map = {
AgentRole.COORDINATOR: ["task_splitter", "agent_selector", "progress_tracker"],
AgentRole.RESEARCHER: ["search_engine", "data_extractor", "source_validator"],
AgentRole.ANALYZER: ["statistical_analyzer", "pattern_detector", "visualization_tool"],
AgentRole.EXECUTOR: ["code_executor", "file_handler", "api_caller"]
}
return tools_map.get(self.role, [])
def process_message(self, message: str, context: Optional[Dict] = None) -> Dict[str, Any]:
"""Process incoming message with context awareness - Synchronous version"""
msg is a message (
role="user",
content=message,
timestamp=datetime.now(),
metadata=context
)
self.memory.append(msg)
prompt = self._generate_contextual_prompt(message, context)
try:
If self.modelReturn
response = self._generate_response_gemini(prompt)
else:
response = self._generate_demo_response(message)
response_msg = Message(
role="assistant",
content=response,
timestamp=datetime.now(),
metadata={"agent_id": self.agent_id, "role": self.role.value}
)
self.memory.append(response_msg)
return {
"agent_id": self.agent_id,
"role": self.role.value,
"response": response,
"capabilities_used": self._analyze_capabilities_used(message),
"next_actions": self._suggest_next_actions(response),
"timestamp": datetime.now().isoformat()
}
Other than Exceptions as follows:Return
logger.error(f"Error processing message: {e}")
return {"error": str(e)}
def _generate_response_gemini(self, prompt: str) -> str:
"""Generate response using Gemini API - Synchronous"""
try:
response = self.model.generate_content(prompt)
Text return for response
Only Exceptions as follows:
logger.error(f"Gemini API error: {e}")
return self._generate_demo_response(prompt)
def _generate_demo_response(self, message: str) -> str:
"""Generate simulated response for demo purposes"""
role_responses = {
AgentRole.COORDINATOR: f"As coordinator, I'll break down the task: '{message[:50]}...' into manageable components and assign them to specialized agents.",
AgentRole.RESEARCHER: f"I'll research information about: '{message[:50]}...' using my data gathering and synthesis capabilities.",
AgentRole.ANALYZER : f"Analyzing the patterns and insights from: '{message[:50]}...' to provide data-driven recommendations.",
AgentRole.EXECUTOR :"I'll execute the necessary actions for: '{message[:50]}...' and validate the results."
}
base_response = role_responses.get(self.role, f"Processing: {message[:50]}...")
time.sleep(0.5)
additional_context = {
AgentRole.COORDINATOR: " I've identified 3 key subtasks and will coordinate their execution across the agent team.",
AgentRole.RESEARCHER: " My research indicates several relevant sources and current trends in this area.",
AgentRole.ANALYZER: " The data shows interesting correlations and actionable insights for decision making.",
AgentRole.EXECUTOR: " I've completed the requested actions and verified the outputs meet quality standards."
}
return base_response + additional_context.get(self.role, "")
def _generate_contextual_prompt(self, message: str, context: Optional[Dict]) -> str:
"""Generate context-aware prompt based on agent role"""
base_prompt = f"""
You are an advanced AI agent with the role: {self.role.value}
Your capabilities: {', '.join(self.context.capabilities)}
Available tools: {', '.join(self.context.tools)}
Conversation context of recent conversations
{self._get_recent_context()}
Current request: {message}
"""
role_instructions = {
AgentRole.COORDINATOR: """
Focus on breaking complicated tasks down and coordination with other agents.
Keep the overall project cohesive. Take into consideration dependencies and priority.
Decompose the tasks and assign agents clearly.
""",
AgentRole.RESEARCHER: """
Verify the source of information and gather accurate data.
Data collection should be comprehensive. Synthesize your findings.
Current trends and reliable sources are important.
""",
AgentRole.ANALYZER: """
Concentrate on patterns recognition, data analysis, and insights generation.
Give evidence-based recommendations and conclusions.
Focus on the key relationships and their implications.
""",
AgentRole.EXECUTOR: """
Implementation and validation of results should be the focus.
Deliver outputs that are clear. Make sure actions are carried out effectively.
Prioritise quality and completion of the execution.
"""
}
return base_prompt + role_instructions.get(self.role, "")
def _get_recent_context(self, limit: int = 3) -> str:
"""Get recent conversation context"""
If you do not have a self-memory
Return to the Homepage "No previous context"
Recent = memory of self[-limit:]
context_str = ""
Recent msgs:
These are the most common context_str+=f"{msg.role}: {msg.content[:100]}...n"
Return context_str
def _analyze_capabilities_used(self, message: str) -> List[str]:
"""Analyze which capabilities were likely used"""
used_capabilities = []
message_lower = message.lower()
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 capability, keywords in capability_keywords.items():
if capability in self.context.capabilities:
If any (keywords in the message_lower case for keywords in the keyword):
used_capabilities.append(capability)
return used_capabilities
def _suggest_next_actions(self, response: str) -> List[str]:
"""Suggest logical next actions based on response"""
You can also try these suggestions []
response_lower = response.lower()
If "need more information" Lower or "research" In response_lower
suggestions.append("delegate_to_researcher")
If "analyze" Lower or "pattern" In response_lower
suggestions.append("delegate_to_analyzer")
If "implement" Lower or "execute" In response_lower
suggestions.append("delegate_to_executor")
If you want to know more about if "coordinate" Lower or "manage" In response_lower
suggestions.append("initiate_multi_agent_collaboration")
If "subtask" Lower or "break down" In response_lower
suggestions.append("task_decomposition_required")
Return suggestions if they are not the ones you wanted ["continue_conversation"]
This AI Paper Introduces MMaDA: A Unified Multimodal Diffusion Model
