Spaces:
Sleeping
Sleeping
File size: 2,668 Bytes
e9aab05 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
import requests
import json
from tinytroupe.agent.mental_faculty import TinyToolUse
from tinytroupe.utils.logger import get_logger
class SequentialThinkingTool(TinyToolUse):
def __init__(self):
super().__init__(tools=[self])
self.url = "https://harvesthealth-sequential-thinking-mcp.hf.space/run"
def process_action(self, agent, action: dict) -> bool:
if action['type'] == 'SEQUENTIAL_THINKING':
logger = get_logger(agent.name)
try:
arguments = json.loads(action['content'])
except json.JSONDecodeError as e:
logger.error(f"MCP Interaction - Invalid JSON in action content: {action['content']}. Error: {e}")
return False
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "sequentialthinking",
"arguments": arguments
}
}
logger.info(f"MCP Interaction - Request: {json.dumps(payload, indent=2)}")
response_json = self.send_thought(payload)
logger.info(f"MCP Interaction - Response: {json.dumps(response_json, indent=2)}")
if response_json and 'result' in response_json and 'content' in response_json['result']:
content_text = response_json['result']['content'][0]['text']
try:
response_data = json.loads(content_text)
agent.think(f"Thought processed. History length: {response_data.get('thoughtHistoryLength')}")
except json.JSONDecodeError:
logger.error(f"MCP Interaction - Could not decode response content: {content_text}")
agent.think("Received a response from the sequential thinking server, but it was not in the expected format.")
return True
return False
def send_thought(self, thought_data: dict):
headers = {'Content-Type': 'application/json'}
try:
response = requests.post(self.url, headers=headers, json=thought_data)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
# Get the logger for the agent that is making the call
# This is a bit of a hack, as we don't have the agent object here.
# We will rely on the caller to log the error.
return {"error": str(e)}
def actions_definitions_prompt(self) -> str:
return ""
def actions_constraints_prompt(self) -> str:
return ""
|