Why Is 'self' Missing from Instance Langchain Tool Methods in Custom CrewAI Agents?

I'm implementing custom CrewAI agents that use langchain tools to interact with a PostgreSQL database. In my setup, the ToolWithConnection class establishes a database connection and defines tool methods such as list_tables and tables_schema. These instance methods are then passed to agents in the CrewAIWrapper. However, when executing the agent tasks, I encounter the error:

PYTHONCopied
I encountered an error while trying to use the tool. This was the error: ToolWithConnection.list_tables() missing 1 required positional argument: 'self'.

Below is a simplified version of my code:

Connection Method:

PYTHONCopied
def connection(x):
    postgres_schm = x["database_schm"]
    postgres_host = x["database_host"]
    postgres_port = x["database_port"]
    postgres_user = x["database_user"]
    postgres_pass = x["database_pass"]
    postgres_dryr = x["database_dryr"]

    if postgres_dryr == "true":
        postgres_pass_dec = postgres_pass
    else:
        postgres_pass_dec = decrypt_string(postgres_pass)

    CONNECTION_STRING = (
        f"postgresql+psycopg2://{postgres_user}:{postgres_pass_dec}@{postgres_host}:{postgres_port}/{postgres_schm}"
    )

    db = SQLDatabase.from_uri(
        CONNECTION_STRING,
        schema=postgres_schm,
        view_support=True
    )
    return db

Tool Class:

PYTHONCopied
class ToolWithConnection:
    def __init__(self, x):
        """Initialize with a connection based on the input 'x'."""
        self.db = connection(x)
        print('>> Initialize db connection ✔\n')

    @tool("list_tables")
    def list_tables(self) -> str:
        """List the available tables in the database."""
        return ListSQLDatabaseTool(db=self.db).invoke("")

    @tool("tables_schema")
    def tables_schema(self, tables: str) -> str:
        """Get the schema of specified tables."""
        tool = InfoSQLDatabaseTool(db=self.db)
        return tool.invoke(tables)

CrewAI Wrapper:

PYTHONCopied
class CrewAIWrapper:
    def __init__(self, x):
        self.x = x
        self.llm_crew = LLM(model='azure/GPT4o')
        tool_instance = ToolWithConnection(self.x)

        self.sql_dev = Agent(
            role="Senior Database Developer",
            goal="Construct and execute PostgreSQL queries based on a query {query}",
            tools=[
                tool_instance.list_tables,  
                tool_instance.tables_schema,
            ],
            allow_delegation=False,
            memory=False,
            verbose=True,
        )

    def kickoff_crew(self):
        extract_data_task = Task(
            description="Extract data that is required for the query {query}.",
            agent=self.sql_dev,
        )

        my_crew = Crew(
            agents=[self.sql_dev],
            tasks=[extract_data_task],
            verbose=True,
        )

        try:
            crew_result = my_crew.kickoff(
                inputs={"query": self.x['question']}
            )
            return crew_result.raw
        except Exception as e:
            print(f"Error during task execution: {str(e)}")

Chain Setup:

PYTHONCopied
class CustomCrewAILLM(LLM):
    model: str
    def __init__(self, model='azure/GPT4o'):
        super().__init__(model=model)
    @property
    def _llm_type(self):
        return 'custom_crew_ai'
    def _call(self, obj: dict,) -> str:
        crew_ai_wrapper = CrewAIWrapper(obj, )
        result = crew_ai_wrapper.kickoff_crew()
        print(type(result))
        # print(result)
        return result
class InputType(BaseModel):
    crewai_input: Optional[str]
    db: Optional[SQLDatabase] = Field(default=None, exclude=True)  # Exclude from schema validation
    database_schm: str
    database_host: str
    database_port: str
    database_user: str
    database_pass: str
    database_name: str
    database_dryr: str
    class Config:
        arbitrary_types_allowed = True  # Allow non-Pydantic types like SQLDatabase
class OutputType(BaseModel):
    crewai_output: Optional[str]
chain = (
    RunnablePassthrough.assign(
        crewai_output=RunnableLambda(lambda x: CustomCrewAILLM()._call(x))
        )
        ).with_types(input_type=InputType,
                     output_type=OutputType
                     )

The Problem:

When running CrewAIWrapper with the provided agents and tools, I encounter the error that indicates the list_tables tool is missing the required self argument.

My Questions:

• Why is the self argument missing when calling list_tables?
• How can I properly pass instance methods (like list_tables) from a class to agents without facing this error?

I appreciate any insights or solutions to ensure that the methods remain properly bound when used as tools in the custom agent setup.