class MCPAgent:
"""Superior MCP Agent with developed capabilities - Jupyter Suitable"""
def __init__(self, agent_id: str, position: AgentRole, api_key: str = None):
self.agent_id = agent_id
self.position = position
self.api_key = api_key
self.reminiscence = []
self.context = AgentContext(
agent_id=agent_id,
position=position,
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("💡 Working in demo mode with simulated responses")
else:
print(f"🎭 Agent {agent_id} working in demo mode")
def _init_capabilities(self) -> Record[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.position, [])
def _init_tools(self) -> Record[str]:
"""Initialize out there instruments based mostly on position"""
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.position, [])
def process_message(self, message: str, context: Non-obligatory[Dict] = None) -> Dict[str, Any]:
"""Course of incoming message with context consciousness - Synchronous model"""
msg = Message(
position="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(
position="assistant",
content material=response,
timestamp=datetime.now(),
metadata={"agent_id": self.agent_id, "position": self.position.worth}
)
self.reminiscence.append(response_msg)
return {
"agent_id": self.agent_id,
"position": self.position.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 = {
AgentRole.COORDINATOR: f"As coordinator, I will break down the duty: '{message[:50]}...' into manageable parts and assign them to specialised brokers.",
AgentRole.RESEARCHER: f"I will analysis details about: '{message[:50]}...' utilizing my information gathering and synthesis capabilities.",
AgentRole.ANALYZER: f"Analyzing the patterns and insights from: '{message[:50]}...' to supply data-driven suggestions.",
AgentRole.EXECUTOR: f"I will execute the required actions for: '{message[:50]}...' and validate the outcomes."
}
base_response = role_responses.get(self.position, f"Processing: {message[:50]}...")
time.sleep(0.5)
additional_context = {
AgentRole.COORDINATOR: " I've recognized 3 key subtasks and can coordinate their execution throughout the agent staff.",
AgentRole.RESEARCHER: " My analysis signifies a number of related sources and present developments on this space.",
AgentRole.ANALYZER: " The information exhibits fascinating correlations and actionable insights for determination making.",
AgentRole.EXECUTOR: " I've accomplished the requested actions and verified the outputs meet high quality requirements."
}
return base_response + additional_context.get(self.position, "")
def _generate_contextual_prompt(self, message: str, context: Non-obligatory[Dict]) -> str:
"""Generate context-aware immediate based mostly on agent position"""
base_prompt = f"""
You might be a sophisticated AI agent with the position: {self.position.worth}
Your capabilities: {', '.be a part of(self.context.capabilities)}
Out there instruments: {', '.be a part of(self.context.instruments)}
Current dialog context:
{self._get_recent_context()}
Present request: {message}
"""
role_instructions = {
AgentRole.COORDINATOR: """
Give attention to breaking down complicated duties, coordinating with different brokers,
and sustaining total undertaking coherence. Contemplate dependencies and priorities.
Present clear activity decomposition and agent assignments.
""",
AgentRole.RESEARCHER: """
Prioritize correct info gathering, supply verification,
and complete information assortment. Synthesize findings clearly.
Give attention to present developments and dependable sources.
""",
AgentRole.ANALYZER: """
Give attention to sample recognition, information interpretation, and perception technology.
Present evidence-based conclusions and actionable suggestions.
Spotlight key correlations and implications.
""",
AgentRole.EXECUTOR: """
Think about sensible implementation, end result validation,
and clear output supply. Guarantee actions are accomplished successfully.
Give attention to high quality and completeness of execution.
"""
}
return base_prompt + role_instructions.get(self.position, "")
def _get_recent_context(self, restrict: int = 3) -> str:
"""Get current dialog context"""
if not self.reminiscence:
return "No earlier context"
current = self.reminiscence[-limit:]
context_str = ""
for msg in current:
context_str += f"{msg.position}: {msg.content material[:100]}...n"
return context_str
def _analyze_capabilities_used(self, message: str) -> Record[str]:
"""Analyze which capabilities have been possible 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) -> Record[str]:
"""Recommend logical subsequent actions based mostly on response"""
options = []
response_lower = response.decrease()
if "want extra info" in response_lower or "analysis" in response_lower:
options.append("delegate_to_researcher")
if "analyze" in response_lower or "sample" in response_lower:
options.append("delegate_to_analyzer")
if "implement" in response_lower or "execute" in response_lower:
options.append("delegate_to_executor")
if "coordinate" in response_lower or "handle" in response_lower:
options.append("initiate_multi_agent_collaboration")
if "subtask" in response_lower or "break down" in response_lower:
options.append("task_decomposition_required")
return options if options else ["continue_conversation"]
