I am attempting to create my first custom tool that I want to leverage a python script I have to retrieve data. I have not managed to get this to work, but I guess my basic question is, “Should I be using a custom tool or can an agent call a python script?” I have validated that the tool being called matches the case/class defined in agents/tasks/crew files.
Custom_tool.py:
from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type
import os
import subprocess
import json
from datetime import datetime
class AzureDevOpsToolInput(BaseModel):
start_date: str = Field(…, description=“Start date (YYYY-MM-DD)”)
end_date: str = Field(…, description=“End date (YYYY-MM-DD)”)
class AzureDevOpsCustomTool(BaseTool):
name: str = “azure_devops_tool”
description: str = “Fetches Azure DevOps work items for a date range”
args_schema: Type[BaseModel] = AzureDevOpsToolInput
def _run(self, start_date: str, end_date: str) -> str:
try:
script_content = '''
import requests
import json
import base64
from datetime import datetime
ADO_ORG = “REMOVED”
ADO_PROJECT = “REMOVED”
ADO_TEAM = “REMOVED”
ADO_AREA_PATH = “REMOVED”
ADO_PAT = os.getenv(“ADO_PAT”)
BASE_URL = f"https://dev.azure.com/{ADO_ORG}/{ADO_PROJECT}/_apis/wit"
auth_token = base64.b64encode(f":{ADO_PAT}“.encode()).decode()
HEADERS = {
“Authorization”: f"Basic {auth_token}”,
“Content-Type”: “application/json”
}
def get_work_items(start_date, end_date):
wiql_query = {
“query”: f"“”
SELECT [System.Id]
FROM WorkItems
WHERE [System.TeamProject] = ‘{ADO_PROJECT}’
AND [System.ChangedDate] >= ‘{start_date}’
AND [System.ChangedDate] <= ‘{end_date}’
“”"
}
response = requests.post(f"{BASE_URL}/wiql?api-version=6.0",
headers=HEADERS,
json=wiql_query)
response.raise_for_status()
return response.json().get(‘workItems’, )
work_items = get_work_items(“{start_date}”, “{end_date}”)
print(json.dumps(work_items, indent=2))
‘’’
current_dir = os.path.dirname(os.path.abspath(file))
script_path = os.path.join(current_dir, “temp_ado_script.py”)
try:
with open(script_path, "w") as f:
f.write(script_content)
result = subprocess.run(
["python", script_path],
capture_output=True,
text=True,
env=os.environ.copy()
)
if result.returncode != 0:
raise Exception(f"Script error: {result.stderr}")
return result.stdout
finally:
if os.path.exists(script_path):
os.remove(script_path)
except Exception as e:
return f"Error: {str(e)}"
Error:
KeyError: ‘azure_devops_tool’
An error occurred while running the crew: Command ‘[‘uv’, ‘run’, ‘run_crew’]’ returned non-zero exit status 1.