Coverage for tinytroupe / tools / sequential_thinking.py: 0%

1import requests 

2import json 

3from tinytroupe.agent.mental_faculty import TinyToolUse 

4from tinytroupe.utils.logger import get_logger 

5 

6class SequentialThinkingTool(TinyToolUse): 

7 def __init__(self): 

8 super().__init__(tools=[self]) 

9 self.url = "https://harvesthealth-sequential-thinking-mcp.hf.space/run" 

10 

11 def process_action(self, agent, action: dict) -> bool: 

12 if action['type'] == 'SEQUENTIAL_THINKING': 

13 logger = get_logger(agent.name) 

14 

15 try: 

16 arguments = json.loads(action['content']) 

17 except json.JSONDecodeError as e: 

18 logger.error(f"MCP Interaction - Invalid JSON in action content: {action['content']}. Error: {e}") 

19 return False 

20 

21 payload = { 

22 "jsonrpc": "2.0", 

23 "id": 1, 

24 "method": "tools/call", 

25 "params": { 

26 "name": "sequentialthinking", 

27 "arguments": arguments 

28 } 

29 } 

30 

31 logger.info(f"MCP Interaction - Request: {json.dumps(payload, indent=2)}") 

32 response_json = self.send_thought(payload) 

33 logger.info(f"MCP Interaction - Response: {json.dumps(response_json, indent=2)}") 

34 

35 if response_json and 'result' in response_json and 'content' in response_json['result']: 

36 content_text = response_json['result']['content'][0]['text'] 

37 try: 

38 response_data = json.loads(content_text) 

39 agent.think(f"Thought processed. History length: {response_data.get('thoughtHistoryLength')}") 

40 except json.JSONDecodeError: 

41 logger.error(f"MCP Interaction - Could not decode response content: {content_text}") 

42 agent.think("Received a response from the sequential thinking server, but it was not in the expected format.") 

43 

44 return True 

45 return False 

46 

47 def send_thought(self, thought_data: dict): 

48 headers = {'Content-Type': 'application/json'} 

49 try: 

50 response = requests.post(self.url, headers=headers, json=thought_data) 

51 response.raise_for_status() 

52 return response.json() 

53 except requests.exceptions.RequestException as e: 

54 # Get the logger for the agent that is making the call 

55 # This is a bit of a hack, as we don't have the agent object here. 

56 # We will rely on the caller to log the error. 

57 return {"error": str(e)} 

58 

59 def actions_definitions_prompt(self) -> str: 

60 return "" 

61 

62 def actions_constraints_prompt(self) -> str: 

63 return ""