Hi Tony
This is the error that blocks me :
Error: the Action Input is not a valid key, value dictionary.
The first agent produces the format I am expecting but the last one struggles with the input.
I have defined a model.py that structure the output exactly as the mongodB schema but still… I am stuck with this
Error: the Action Input is not a valid key, value dictionary.
# Agent: MongodB Senior Administrator
## Thought: Thought: I need to ensure that the data is structured correctly for the MongoDB update. I will format the input data according to the specified criteria.
## Using tool: Update_MongodB
## Tool Input:
"{\"input_data\": {\"Id\": \"4ilqI0k\", \"days\": [{\"event_date\": \"2024-11-22\", \"events\": [{\"folder\": \"Shopping\", \"eventType\": \"Delivery\", \"title\": \"******************** *********appointment.\", \"address\": null, \"attendees\": [\"Axel\"], \"reminder\": {\"enabled\": true, \"remind_at\": \"2024-12-02T08:00:00.000Z\"}, \"source\": null, \"notification_date\": null}]}]}}"
## Tool Output:
Error: the Action Input is not a valid key, value dictionary.
My Crew
@crew
def crew(self) -> Crew:
"""Creates the Calendar crew"""
return Crew(
agents=[
self.calendar_manager_assistant(),
self.senior_personal_assistant(),
self.mongodb_administrator(),
],
tasks=[
self.collect_calendar_events(),
self.compile_calendar_events(),
self.Update_Calendar_to_MongodB(),
],
process=Process.sequential,
verbose=True,
#memory=True,
callback_manager=self.callback_manager
)
my tasks
@task
def collect_calendar_events(self) -> Task:
"""Creates the grades collection task"""
return Task(
config=self.tasks['collect_calendar_events'],
agent=self.calendar_manager_assistant(),
#output_pydantic=CalendarEvent
)
@task
def compile_calendar_events(self) -> Task:
"""Creates the grades collection task"""
return Task(
config=self.tasks['compile_calendar_events'],
agent=self.senior_personal_assistant(),
context=[self.collect_calendar_events()],
output_pydantic=CalendarEvent
)
@task
def Update_Calendar_to_MongodB(self) -> Task:
"""Updates MongoDB with grades"""
return Task(
config=self.tasks['Update_Calendar_to_MongodB'],
agent=self.mongodb_administrator(),
context=[self.compile_calendar_events()],
#output_pydantic=CalendarEvent
)
The tool
tools/Update_MongodB.py
from crewai_tools import BaseTool
from pymongo import MongoClient, errors
from typing import Dict, Any
from pydantic import BaseModel, Field, PrivateAttr
import os
from datetime import datetime, timezone
import logging
import json
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class Update_MongodB(BaseTool):
name: str = "Update_MongodB"
description: str = "Tool to update MongoDB database "
uniqueid: str = Field(..., description="The unique ID for database operations")
_client: Any = PrivateAttr(default=None)
_db: Any = PrivateAttr(default=None)
_collection: Any = PrivateAttr(default=None)
def __init__(self, **data):
super().__init__(**data)
try:
mongo_uri = os.getenv("*****")
if not mongo_uri:
raise ValueError("*******")
self._client = MongoClient(
mongo_uri,
serverSelectionTimeoutMS=5000,
connectTimeoutMS=5000,
socketTimeoutMS=5000
)
self._client.admin.command('ping')
self._db = self._client['*****']
self._collection = self._db['****']
except Exception as e:
print(f"ERROR in initialization: {str(e)}")
raise
def _standardize_data(self, data: Any) -> Dict:
"""Standardize input data to match schema"""
try:
# Convert string to dict if needed
if isinstance(data, str):
data = json.loads(data)
# Handle nested data structure
if isinstance(data, dict):
if 'data' in data:
data = data['data']
if 'description' in data:
data = data['description']
return self._standardize_event_structure(data)
except Exception as e:
print(f"ERROR in _standardize_data: {str(e)}")
raise
def _standardize_event_structure(self, data: Dict) -> Dict:
"""Helper method to standardize event structure"""
try:
standardized_data = {
"clerkId": data.get('clerkId', self.clerk_id),
"days": []
}
for day in data.get('days', []):
standardized_day = {
"date": day.get('date') or day.get('event_date'),
"events": []
}
for event in day.get('events', []):
standardized_event = self._standardize_event(event)
if standardized_event:
standardized_day['events'].append(standardized_event)
if standardized_day['events']:
standardized_data['days'].append(standardized_day)
return standardized_data
except Exception as e:
raise
def _run(self, input_data: Any) -> Dict[str, Any]:
"""Execute the tool's primary function"""
try:
# Convert input to dictionary if it's a string
if isinstance(input_data, str):
print("Attempting to parse JSON string...")
try:
data = json.loads(input_data)
except json.JSONDecodeError as e:
raise ValueError("Invalid JSON input")
else:
data = input_data
print("\nExtracting nested data...")
# Extract nested data if needed
if isinstance(data, dict):
if 'input' in data:
data = data['input']
if 'data' in data:
data = data['data']
if 'description' in data:
data = data['description']
print(f"Extracted data: {data}")
# Validate required fields
if not isinstance(data, dict):
raise ValueError(f"Expected dictionary input, got {type(data)}")
if 'clerkId' not in data:
data['clerkId'] = self.clerk_id
if 'days' not in data or not isinstance(data['days'], list):
raise ValueError("Input must contain 'days' array")
# Update MongoDB
update_result = self._update_calendar_structure(data)
return {
"status": "success",
"message": "Successfully updated calendar events",
"data": {
"clerkId": self.clerk_id,
"update_result": update_result
}
}
except Exception as e:
print(f"\nERROR in _run method: {str(e)}")
print(f"Error type: {type(e)}")
print(f"Error details: {str(e)}")
return {
"status": "error",
"message": str(e),
"data": None
}
def __del__(self):
"""Cleanup MongoDB connection"""
try:
if hasattr(self, '_client') and self._client:
self._client.close()
logger.info("MongoDB connection closed")
except Exception as e:
logger.error(f"Error closing MongoDB connection: {str(e)}")
Thanks a lot for your support.
Kind Regards
Arnaud
Your help would be greatly appreciated.