TypeError: Flow.kickoff() takes 1 positional argument but 2 were given

General
, ,
1 
import os
from datetime import datetime
from pydantic import BaseModel, Field
from crewai.flow.flow import Flow, listen, start
# from tools.tools import listener  
from pydantic_types import CalendarDetails
from crews.calendar_utils_crew.calendar_utils_crew import CalendarUtilsCrew 



# Define State for the Flow
class EmailProcessingState(BaseModel):
    email: str = None
    thread_id: str = None
    sender: str = None
    cc_emails: list = None
    # calendar_details: CalendarDetails = None  
    # processing_complete: bool = False

# Define the Flow class for CalendarUtils
class CalendarSchedulingFlow(Flow[EmailProcessingState]):
    
    @start()
    def start_calendar_utils(self):
        print("inside  start flow...")
        crew_inputs = {
            "sender_mail": self.state.sender,
            "cc_emails": self.state.cc_emails,
            "thread_id": self.state.thread_id,
            "message": self.state.email
        }
        print(crew_inputs)
        result = CalendarUtilsCrew().crew().kickoff(inputs=crew_inputs)
        print(result)
        return result

    @listen("start_calendar_utils")
    def process_calendar_details(self, calendar_output: CalendarDetails):
        self.update_state(calendar_details=calendar_output, processing_complete=True) 
        print('calendar_output',calendar_output)
        print("Processed Calendar Details:")
        print("Timezone:", calendar_output.timezone)
        print("Current Datetime:", calendar_output.current_datetime)
        print("Calendar IDs:", calendar_output.calendar_ids)


def kickoff(email_inputs):
    print('kickoff........')
    print(email_inputs)
    calendar_flow = CalendarSchedulingFlow()
    calendar_flow.kickoff(email_inputs)




if __name__ == "__main__":
    print("Starting Gmail Listener...")
    # listener.wait_forever()
    
    
    email_inputs = {'email': 'Hey gia,\r\nplease book a meeting for 1 hour tomm?\r\n', 'thread_id': '1936fbacfa9ea0c3', 'sender': 'rakshit gupta <guptarakshit2000@gmail.com>', 'cc_emails': [], 'subject': 'book meeting'}
    kickoff(email_inputs)

Traceback (most recent call last):
File “/Users/rakshitgupta/Desktop/AGEIA/02 Project/Agiea-experiments/email_flow/main.py”, line 103, in
kickoff(email_inputs)
File “/Users/rakshitgupta/Desktop/AGEIA/02 Project/Agiea-experiments/email_flow/main.py”, line 54, in kickoff
calendar_flow.kickoff(email_inputs)
TypeError: Flow.kickoff() takes 1 positional argument but 2 were given

2

Hey @rakshit_gupta, try this. I hope this should work

import os
from datetime import datetime
from pydantic import BaseModel, Field
from crewai.flow.flow import Flow, listen, start
from pydantic_types import CalendarDetails
from crews.calendar_utils_crew.calendar_utils_crew import CalendarUtilsCrew


# # Define State for the Flow
# class EmailProcessingState(BaseModel):
#     email: str = None
#     thread_id: str = None
#     sender: str = None
#     cc_emails: list = None
#     # calendar_details: CalendarDetails = None
#     # processing_complete: bool = False


# Define the Flow class for CalendarUtils
class CalendarSchedulingFlow(Flow):
    def __init__(self, input_dict):
        super().__init__()
        self.state["email"] = input_dict["email"]
        self.state["thread_id"] = input_dict["thread_id"]
        self.state["sender"] = input_dict["sender"]
        self.state["cc_emails"] = input_dict["cc_emails"]
        self.state["subject"] = input_dict["subject"]
        self.state["calendar_details"] = None
        self.state["processing_complete"] = False

    @start()
    def start_calendar_utils(self):
        print("inside  start flow...")
        crew_inputs = {
            "sender_mail": self.state["sender"],
            "cc_emails": self.state["cc_emails"],
            "thread_id": self.state["thread_id"],
            "message": self.state["email"]
        }
        print(crew_inputs)
        calendar_output = CalendarUtilsCrew().crew().kickoff(inputs=crew_inputs)
        print(calendar_output)
        return calendar_output

    @listen("start_calendar_utils")
    def process_calendar_details(self, calendar_output):
        self.state["calendar_details"] = calendar_output
        self.state["processing_complete"] = True
        print('calendar_output', calendar_output)
        print("Processed Calendar Details:")
        print("Timezone:", calendar_output.timezone)
        print("Current Datetime:", calendar_output.current_datetime)
        print("Calendar IDs:", calendar_output.calendar_ids)


def kickoff(email_inputs):
    print('kickoff........')
    print(email_inputs)
    calendar_flow = CalendarSchedulingFlow(email_inputs)
    calendar_flow.kickoff()


if __name__ == "__main__":
    print("Starting Gmail Listener...")
    # listener.wait_forever()

    email_inputs = {'email': 'Hey gia,\r\nplease book a meeting for 1 hour tomm?\r\n', 'thread_id': '1936fbacfa9ea0c3',
                    'sender': 'rakshit gupta <guptarakshit2000@gmail.com>', 'cc_emails': [], 'subject': 'book meeting'}
    kickoff(email_inputs)
3

Thanks for the solution! But with this then i will switch it in Unstructured State Management, I want to do Structured way, there documentation shows it is possible but it is not working.