Error: Input should be a valid dictionary or instance of BaseTool

CrewAI Community Support
1

Screenshot 2025-03-18 112107
Screenshot 2025-03-18 1121071474×179 8.43 KB

Hi, I am facing this issue when working with custom tools
Code Snippet:

class SearchCoinGeckoAPIInput(BaseModel):
query: str = Field(…, description=“Query string to search for coin information.”)

class CryptoTools(BaseTool):
name: str = “Search CoinGecko API”
description: str = “Searches CoinGecko Pro API for coin information using a query string.”
args_schema: Type[BaseModel] = SearchCoinGeckoAPIInput

coingecko_api_key: str = os.getenv("COINGECKO_API_KEY", "")
if not coingecko_api_key:
    raise ValueError("COINGECKO_API_KEY is missing! Set it in environment variables.")

base_headers: Dict[str, str] = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
    'x-cg-pro-api-key': coingecko_api_key
}

def _run(self, query: str) → str:
“”“Search CoinGecko Pro API for coin information using a query string.”“”
try:
url = f"https://pro-api.coingecko.com/api/v3/search?query={query}"
response = requests.get(url, headers=self.base_headers)
response.raise_for_status()

        data = response.json()
        coins = data.get('coins', [])

        if not coins:
            return {
                "success": False,
                "message": f"No results found for '{query}' on CoinGecko Pro"
            }

        best_match = coins[0]
        return {
            "success": True,
            "coin_id": best_match.get('id'),
            "name": best_match.get('name'),
            "symbol": best_match.get('symbol', '').upper(),
            "market_cap_rank": best_match.get('market_cap_rank')
        }

    except requests.exceptions.HTTPError as e:
        return {
            "success": False,
            "message": f"HTTP error: {e.response.status_code} - {e.response.text}"
        }
    except requests.exceptions.RequestException as e:
        return {
            "success": False,
            "message": f"Request error: {str(e)}"
        }
    except Exception as e:
        return {
            "success": False,
            "message": f"Unexpected error: {str(e)}"
        }

Even after extending the BaseTool class the error still remains unresolved.

Anyone who has faced similar issue, please guide.

2

How are you instantiating your tool and passing it to the agent?

Can you share that code too?

3

from crewai import Agent, Task
from langchain_openai import ChatOpenAI
from tools.coin_id_tool import CryptoTools
from .base_agent import BaseAgent
from utils.llm_provider import LLMProvider
import os

class AddressExtractorAgent(BaseAgent):
def init(self):
# Initialize the LLM provider
self.llm_provider = LLMProvider()
self.crypto_tools = CryptoTools()

def get_agent(self) -> Agent:
    
    return Agent(
        role="Cryptocurrency Information Researcher",
        goal="Find accurate coin IDs and information for cryptocurrencies",
        backstory="""You are an expert in cryptocurrency data extraction and identification. Your specialty 
        is analyzing search results to identify the exact cryptocurrency being requested, even when there 
        are multiple similar options. You understand the importance of market cap rank, trading volume, 
        and other metrics to determine the most relevant cryptocurrency. You're meticulous 
        about data accuracy and always verify information before proceeding.""",
        verbose=True,
        allow_delegation=False,
        tools=[
            self.crypto_tools
        ],
        llm=self.llm_provider.llm
    )

def process_query(self, query: str) -> dict:
    """
    Process a query by running it through the agent and returning a structured response.
    
    Args:
        query (str): The query to process, typically in the format "Find information about {coin_name}"
        
    Returns:
        dict: A dictionary containing the status and either the result or error message
    """
    try:
        # Create a new agent instance
        agent = self.get_agent()
        
        # Create a task for the agent
        task = Task(
            description=query,
            expected_output="JSON string containing coin_id and any found addresses"
        )
        
        # Execute the task using the agent
        result = agent.execute_task(task)
        return {
            "status": "success",
            "result": result
        }
    except Exception as e:
        return {
            "status": "error",
            "message": str(e)
        }

Keep the old class for backward compatibility if needed

class AddressAndIdExtractor:
@staticmethod
def create():
return AddressExtractorAgent().get_agent()

this is the code for agent

4

Try this code for the tool. Basically add init

import os
import requests
from typing import Dict, Type
from pydantic import BaseModel, Field
from crewai.tools.base_tool import BaseTool

class SearchCoinGeckoAPIInput(BaseModel):
    query: str = Field(..., description="Query string to search for coin information.")

class CryptoTools(BaseTool):
    name: str = "get_crypto_for"
    description: str = "Searches CoinGecko Pro API for coin information using a query string."
    args_schema: Type[BaseModel] = SearchCoinGeckoAPIInput
    
    def __init__(self, **data):
        super().__init__(**data)
        self.coingecko_api_key = os.getenv("COINGECKO_API_KEY", "")
        if not self.coingecko_api_key:
            raise ValueError("COINGECKO_API_KEY is missing! Set it in environment variables.")
        
        self.base_headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'x-cg-pro-api-key': self.coingecko_api_key
        }
    
    def _run(self, query: str) -> str:
        """Search CoinGecko Pro API for coin information using a query string."""
        try:
            url = f"https://pro-api.coingecko.com/api/v3/search?query={query}"
            response = requests.get(url, headers=self.base_headers)
            response.raise_for_status()

            data = response.json()
            coins = data.get('coins', [])

            if not coins:
                return {
                    "success": False,
                    "message": f"No results found for '{query}' on CoinGecko Pro"
                }

            best_match = coins[0]
            return {
                "success": True,
                "coin_id": best_match.get('id'),
                "name": best_match.get('name'),
                "symbol": best_match.get('symbol', '').upper(),
                "market_cap_rank": best_match.get('market_cap_rank')
            }

        except requests.exceptions.HTTPError as e:
            return {
                "success": False,
                "message": f"HTTP error: {e.response.status_code} - {e.response.text}"
            }
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "message": f"Request error: {str(e)}"
            }
        except Exception as e:
            return {
                "success": False,
                "message": f"Unexpected error: {str(e)}"
            }
5

I have tried that too, tried again as you suggested, but still I am getting the same error.

6

I can share main.py code too, if it makes it easier for figuring out the issue

7

With the code I shared you should be getting the following errors:
ValueError: "CryptoTools" object has no field "coingecko_api_key"
ValueError: "CryptoTools" object has no field "base_headers"

The code below works for me when I run in a crewai app

# src/chat_demo/tools/custom_tool.py
import os
import requests
from typing import Dict, Type
from pydantic import BaseModel, Field
from crewai.tools.base_tool import BaseTool


class SearchCoinGeckoAPIInput(BaseModel):
    query: str = Field(..., description="Query string to search for coin information.")


class CryptoTools(BaseTool):
    name: str = "get_crypto_for"
    description: str = (
        "Searches CoinGecko Pro API for coin information using a query string."
    )
    args_schema: Type[BaseModel] = SearchCoinGeckoAPIInput
    coingecko_api_key: str = ""
    base_headers: Dict = {}

    def __init__(self, **data):
        super().__init__(**data)
        self.coingecko_api_key = os.getenv("COINGECKO_API_KEY", "")
        if not self.coingecko_api_key:
            raise ValueError(
                "COINGECKO_API_KEY is missing! Set it in environment variables."
            )

        self.base_headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "x-cg-pro-api-key": self.coingecko_api_key,
        }

    def _run(self, query: str) -> str:
        """Search CoinGecko Pro API for coin information using a query string."""
        try:
            url = f"https://pro-api.coingecko.com/api/v3/search?query={query}"
            response = requests.get(url, headers=self.base_headers)
            response.raise_for_status()

            data = response.json()
            coins = data.get("coins", [])

            if not coins:
                return {
                    "success": False,
                    "message": f"No results found for '{query}' on CoinGecko Pro",
                }

            best_match = coins[0]
            return {
                "success": True,
                "coin_id": best_match.get("id"),
                "name": best_match.get("name"),
                "symbol": best_match.get("symbol", "").upper(),
                "market_cap_rank": best_match.get("market_cap_rank"),
            }

        except requests.exceptions.HTTPError as e:
            return {
                "success": False,
                "message": f"HTTP error: {e.response.status_code} - {e.response.text}",
            }
        except requests.exceptions.RequestException as e:
            return {"success": False, "message": f"Request error: {str(e)}"}
        except Exception as e:
            return {"success": False, "message": f"Unexpected error: {str(e)}"}

and

# src/chat_demo/crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task

from .tools.custom_tool import CryptoTools

crypto_tools = CryptoTools()


@CrewBase
class ChatDemo:
    """ChatDemo crew"""

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config["researcher"], tools=[crypto_tools], verbose=True
        )

    @agent
    def reporting_analyst(self) -> Agent:
        return Agent(config=self.agents_config["reporting_analyst"], verbose=True)

    @task
    def research_task(self) -> Task:
        return Task(
            config=self.tasks_config["research_task"],
        )

    @task
    def reporting_task(self) -> Task:
        return Task(config=self.tasks_config["reporting_task"], output_file="report.md")

    @crew
    def crew(self) -> Crew:
        """Creates the ChatDemo crew"""

        return Crew(
            agents=self.agents, 
            tasks=self.tasks, 
            process=Process.sequential,
            verbose=True,
            # process=Process.hierarchical,
        )

If this code doesn’t work then there is probably something wrong with your app. You can push it into a gihub repo and I will take a look at it.

REMEMBER: you need to add your COINGECKO_API_KEY in your .env file

8

still getting the same issue, I’ll try pushing it to a repo. can you give me your email address so I can send you an invite to the repo

9

brizdigital@gmail.com

10

please look into coin_id_tool and coin_id_agent. If there are any other issues in the code, then also let me know as well. Thanks a lot

11

you must have received an invitation

12

Why are you not using the standard CrewAI project structure? Having straight forward crews might help.

I’ll take a look when I get a bit of free time later.

13

What do you mean by standard crewAI structure? Because we followed the official documentation when coding this. Also, it would be helpful if you could take a look soon as I have a deadline coming up. Thanks a lot.