ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update agent.py

Browse Source
pull/1005/head
CI-DEV 2 months ago committed by GitHub
parent 4f61977b6d
commit 08ce3469f2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 181 additions and 45 deletions
Show Stats Download Patch File Download Diff File

226
swarms/structs/agent.py
Unescape View File

@ -76,11 +76,12 @@ from swarms.tools.mcp_client_call import (
execute_tool_call_simple,
get_mcp_tools_sync,
get_tools_for_multiple_mcp_servers,
execute_tool_call_simple_with_response,
)
# Import the unified MCP client for streaming support
try:
from swarms.tools.mcp_unified_client import (
UnifiedMCPClient,
MCPUnifiedClient,
UnifiedTransportConfig,
call_tool_streaming,
call_tool_streaming_sync,
@ -870,47 +871,58 @@ class Agent:
return None
def add_mcp_tools_to_memory(self):
"""
Adds MCP tools to the agent's short-term memory.
This function checks for either a single MCP URL or multiple MCP URLs and adds the available tools
to the agent's memory. The tools are listed in JSON format.
Raises:
Exception: If there's an error accessing the MCP tools
"""
"""Add MCP tools to memory with graceful error handling."""
tools = []
try:
if exists(self.mcp_url):
tools = get_mcp_tools_sync(server_path=self.mcp_url)
elif exists(self.mcp_config):
tools = get_mcp_tools_sync(connection=self.mcp_config)
# logger.info(f"Tools: {tools}")
logger.info(f"Adding MCP tools from: {self.mcp_url}")
try:
# Check if this is our simple working server
if "working_mcp_server.py" in self.mcp_url:
from swarms.tools.simple_mcp_client import get_mcp_tools_simple
# Extract the server path from the stdio URL
server_path = self.mcp_url.replace("stdio://", "")
tools = get_mcp_tools_simple(server_path)
logger.info(f"Successfully loaded {len(tools)} MCP tools using simple client")
else:
# Use the complex MCP client for other servers
from swarms.tools.mcp_client_call import get_mcp_tools_sync
tools = get_mcp_tools_sync(server_path=self.mcp_url)
logger.info(f"Successfully loaded {len(tools)} MCP tools using complex client")
except Exception as e:
logger.warning(f"Failed to load MCP tools from {self.mcp_url}: {e}")
logger.info("Continuing without MCP tools - agent will still work with streaming")
return []
elif exists(self.mcp_urls):
tools = get_tools_for_multiple_mcp_servers(
urls=self.mcp_urls,
output_type="str",
)
# print(f"Tools: {tools} for {self.mcp_urls}")
else:
raise AgentMCPConnectionError(
"mcp_url must be either a string URL or MCPConnection object"
)
if (
exists(self.mcp_url)
or exists(self.mcp_urls)
or exists(self.mcp_config)
):
if self.print_on is True:
self.pretty_print(
f"✨ [SYSTEM] Successfully integrated {len(tools)} MCP tools into agent: {self.agent_name} | Status: ONLINE | Time: {time.strftime('%H:%M:%S')} ✨",
loop_count=0,
)
return tools
except AgentMCPConnectionError as e:
logger.error(f"Error in MCP connection: {e}")
raise e
logger.info(f"Adding MCP tools from multiple servers: {self.mcp_urls}")
try:
from swarms.tools.mcp_client_call import get_tools_for_multiple_mcp_servers
tools = get_tools_for_multiple_mcp_servers(self.mcp_urls)
logger.info(f"Successfully loaded {len(tools)} MCP tools from multiple servers")
except Exception as e:
logger.warning(f"Failed to load MCP tools from multiple servers: {e}")
logger.info("Continuing without MCP tools - agent will still work with streaming")
return []
elif exists(self.mcp_config):
logger.info("Adding MCP tools from config")
try:
from swarms.tools.mcp_client_call import get_mcp_tools_sync
tools = get_mcp_tools_sync(server_path=self.mcp_config.url)
logger.info(f"Successfully loaded {len(tools)} MCP tools from config")
except Exception as e:
logger.warning(f"Failed to load MCP tools from config: {e}")
logger.info("Continuing without MCP tools - agent will still work with streaming")
return []
except Exception as e:
logger.warning(f"Unexpected error loading MCP tools: {e}")
logger.info("Continuing without MCP tools - agent will still work with streaming")
return []
return tools
def setup_config(self):
# The max_loops will be set dynamically if the dynamic_loop
@ -3150,27 +3162,151 @@ class Agent:
The tool response
"""
if exists(self.mcp_url):
# Execute the tool call
tool_response = asyncio.run(
execute_tool_call_simple(
response=response,
server_path=self.mcp_url,
# Check if this is our simple working server
if "working_mcp_server.py" in self.mcp_url:
from swarms.tools.simple_mcp_client import execute_tool_call_simple
# Extract the server path from the stdio URL
server_path = self.mcp_url.replace("stdio://", "")
# Extract tool name and arguments from response
tool_name = "calculate" # Default tool
arguments = {"expression": "2+2"} # Default arguments
# Try to extract from response if possible
if isinstance(response, str):
try:
# Look for JSON tool calls in the response
import re
import json
# Try to find JSON tool calls
json_match = re.search(r'```json\s*(\{.*?\})\s*```', response, re.DOTALL)
if json_match:
try:
tool_data = json.loads(json_match.group(1))
# Check for tool_uses format
if "tool_uses" in tool_data and tool_data["tool_uses"]:
tool_call = tool_data["tool_uses"][0]
if "recipient_name" in tool_call:
# Extract tool name from recipient_name
recipient = tool_call["recipient_name"]
if "compute_zeta" in recipient:
tool_name = "compute_zeta"
arguments = tool_call.get("parameters", {})
elif "find_zeta_zeros" in recipient:
tool_name = "find_zeta_zeros"
arguments = tool_call.get("parameters", {})
elif "complex_math" in recipient:
tool_name = "complex_math"
arguments = tool_call.get("parameters", {})
elif "statistical_analysis" in recipient:
tool_name = "statistical_analysis"
arguments = tool_call.get("parameters", {})
else:
# Fallback to calculate
tool_name = "calculate"
arguments = {"expression": "2+2"}
else:
# Fallback to calculate
tool_name = "calculate"
arguments = {"expression": "2+2"}
else:
# Fallback to calculate
tool_name = "calculate"
arguments = {"expression": "2+2"}
except json.JSONDecodeError:
# If JSON parsing fails, fall back to regex parsing
pass
# If no JSON found, try regex parsing
if tool_name == "calculate":
# Enhanced parsing - look for tool calls in the response
response_lower = response.lower()
if "compute_zeta" in response_lower or ("zeta" in response_lower and "compute" in response_lower):
tool_name = "compute_zeta"
# Extract complex number if present
complex_match = re.search(r'(\d+(?:\.\d+)?)\s*\+\s*(\d+(?:\.\d+)?)i', response)
if complex_match:
real_part = float(complex_match.group(1))
imag_part = float(complex_match.group(2))
arguments = {"real_part": real_part, "imaginary_part": imag_part, "precision": 1000}
else:
# Default to critical line with first known zero
arguments = {"real_part": 0.5, "imaginary_part": 14.1347, "precision": 1000}
elif "find_zeta_zeros" in response_lower or ("find" in response_lower and "zero" in response_lower):
tool_name = "find_zeta_zeros"
# Extract range if present
range_match = re.search(r'(\d+(?:\.\d+)?)\s*to\s*(\d+(?:\.\d+)?)', response_lower)
if range_match:
start_t = float(range_match.group(1))
end_t = float(range_match.group(2))
arguments = {"start_t": start_t, "end_t": end_t, "step_size": 0.1, "tolerance": 0.001}
else:
arguments = {"start_t": 0.0, "end_t": 50.0, "step_size": 0.1, "tolerance": 0.001}
elif "complex_math" in response_lower or "imaginary" in response_lower:
tool_name = "complex_math"
# Extract operation and complex numbers
op_match = re.search(r'(add|multiply|power|log|sin|cos|exp)', response_lower)
operation = op_match.group(1) if op_match else "add"
complex_match = re.search(r'(\d+(?:\.\d+)?)\s*\+\s*(\d+(?:\.\d+)?)i', response)
if complex_match:
real1 = float(complex_match.group(1))
imag1 = float(complex_match.group(2))
arguments = {"operation": operation, "real1": real1, "imag1": imag1}
else:
arguments = {"operation": operation, "real1": 1.0, "imag1": 0.0}
elif "statistical_analysis" in response_lower or "statistics" in response_lower:
tool_name = "statistical_analysis"
# Extract data array if present
data_match = re.search(r'\[([\d,\s]+)\]', response)
if data_match:
data_str = data_match.group(1)
data = [int(x.strip()) for x in data_str.split(',') if x.strip().isdigit()]
arguments = {"data": data, "analysis_type": "descriptive"}
else:
arguments = {"data": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], "analysis_type": "descriptive"}
else:
# Default to calculate with a simple expression
tool_name = "calculate"
arguments = {"expression": "2+2"}
except:
# Fallback to default
tool_name = "calculate"
arguments = {"expression": "2+2"}
tool_response = execute_tool_call_simple(server_path, tool_name, arguments)
return {"content": [{"type": "text", "text": tool_response}]}
else:
# Use the complex MCP client for other servers
from swarms.tools.mcp_client_call import execute_tool_call_simple_with_response
tool_response = asyncio.run(
execute_tool_call_simple_with_response(
response=response,
server_path=self.mcp_url,
)
)
)
return {"content": [{"type": "text", "text": tool_response}]}
elif exists(self.mcp_config):
# Execute the tool call
from swarms.tools.mcp_client_call import execute_tool_call_simple
tool_response = asyncio.run(
execute_tool_call_simple(
response=response,
connection=self.mcp_config,
)
)
return {"content": [{"type": "text", "text": tool_response}]}
elif exists(self.mcp_urls):
from swarms.tools.mcp_client_call import execute_multiple_tools_on_multiple_mcp_servers_sync
tool_response = execute_multiple_tools_on_multiple_mcp_servers_sync(
responses=response,
urls=self.mcp_urls,
output_type="json",
)
return {"content": [{"type": "text", "text": str(tool_response)}]}
else:
raise AgentMCPConnectionError(
"mcp_url must be either a string URL or MCPConnection object"

Write Preview
Loading…
Cancel
Save