Openai-agents tools

Community Support
1 
        agent = Agent(
            name="Test Data Agent",
            instructions="""
            You are a Test Data Processing Agent that helps users manage test data.
            
            You can:
            1. Read test data from CSV files
            2. Validate if the data is complete
            3. Generate synthetic data for missing fields
            4. Store the processed data in a local cache
            
            When a user asks you to process a CSV file:
            1. First use read_csv_file to analyze the CSV
            2. Check if there are any missing values that need to be filled
            3. Use generate_synthetic_data to fill missing values and/or generate additional synthetic data
            4. Store the processed data using store_test_data
            5. Provide a summary of what you did and the results
            
            Always be helpful and explain your reasoning clearly.
            """,
            model="openai/gpt-4o-(US)",
            tools=[read_csv_file, generate_synthetic_data, store_test_data]
        )

Above is my agent definition. I have configured a read_csv_file tool that reads data from a file

@function_tool
async def read_csv_file(
    ctx: RunContextWrapper[Any], 
    file_path: str,
    sample_rows: int = 5
) -> str:
    """
    Read a CSV file and return the first few rows as a sample.
    
    Args:
        file_path: Path to the CSV file
        sample_rows: Number of rows to sample (default: 5)
        
    Returns:
        JSON string containing metadata and sample data from the CSV
    """
    try:

        print(f"Reading CSV file: {file_path}")
        # Read the CSV file
        df = pd.read_csv(file_path)
        
        # Get metadata
        total_rows = len(df)
        columns = list(df.columns)
        
        # Get a sample of the data (first few rows)
        sample_data = df.head(sample_rows).to_dict(orient='records')
        print(f"Sample data: {sample_data}")
        
        # Calculate missing values per column
        missing_values = df.isna().sum().to_dict()
        print(f"Missing values: {missing_values}")        
        
        # Return the data as a JSON string
        result = {
            "file_path": file_path,
            "total_rows": total_rows,
            "columns": columns,
            "sample_data": sample_data,
            "missing_values": {str(k): int(v) for k, v in missing_values.items()}
        }
        print(f"Result: {result}")

        # Remove the ctx.send_text calls that aren't compatible with Temporal
        # Just return the result directly
        return json.dumps(result, indent=2)
    
    except Exception as e:
        error_msg = f"Error reading CSV file: {str(e)}"
        # Don't use send_text in Temporal workflows
        return json.dumps({"error": error_msg})

I get the following error

{
      "role": "assistant",
      "tool_calls": [
        {
          "id": "call_CIoJg4qTbfTvRilHXZLY1m2k",
          "type": "function",
          "function": {
            "name": "read_csv_file",
            "arguments": "{\"file_path\":\"/Users/xyz/Documents/GitHub/yoda/uploads/incred-test-data - Sheet1 (1).csv\",\"sample_rows\":5}"
          }
        }
      ]
    },
    {
      "role": "tool",
      "tool_call_id": "call_CIoJg4qTbfTvRilHXZLY1m2k",
      "content": "{\"error\": \"Error reading CSV file: Cannot access os.path.expanduser from inside a workflow. If this is code from a module not used in a workflow or known to only be used deterministically from a workflow, mark the import as pass through.\"}"
    }

Any guidance is appreciated.

2

Temporal workflows are not allowed to perform any external IO directly. They must use activities to do this. Change Annotation on read_csv_file from @function_tool to @activity.defn. And use temporal_agents.workflow.activity_as_tool adapter to expose a Temporal activity as a tool to the Agent. Don’t forget to register the activity with the Worker.

        agent = Agent(
            name="Test Data Agent",
            instructions="""
            You are a Test Data Processing Agent that helps users manage test data.
            
            You can:
            1. Read test data from CSV files
            2. Validate if the data is complete
            3. Generate synthetic data for missing fields
            4. Store the processed data in a local cache
            
            When a user asks you to process a CSV file:
            1. First use read_csv_file to analyze the CSV
            2. Check if there are any missing values that need to be filled
            3. Use generate_synthetic_data to fill missing values and/or generate additional synthetic data
            4. Store the processed data using store_test_data
            5. Provide a summary of what you did and the results
            
            Always be helpful and explain your reasoning clearly.
            """,
            model="openai/gpt-4o-(US)",
            tools=[temporal_agents.workflow.activity_as_tool(read_csv_file), 
              temporal_agents.workflow.activity_as_tool(generate_synthetic_data), 
              temporal_agents.workflow.activity_as_tool(store_test_data),
            ]
        )


@ activity.defn
async def read_csv_file(
    file_path: str,
    sample_rows: int = 5
) -> str:
    """
    Read a CSV file and return the first few rows as a sample.
    
    Args:
        file_path: Path to the CSV file
        sample_rows: Number of rows to sample (default: 5)
        
    Returns:
        JSON string containing metadata and sample data from the CSV
    """
    try:

        print(f"Reading CSV file: {file_path}")
        # Read the CSV file
        df = pd.read_csv(file_path)
        
        # Get metadata
        total_rows = len(df)
        columns = list(df.columns)
        
        # Get a sample of the data (first few rows)
        sample_data = df.head(sample_rows).to_dict(orient='records')
        print(f"Sample data: {sample_data}")
        
        # Calculate missing values per column
        missing_values = df.isna().sum().to_dict()
        print(f"Missing values: {missing_values}")        
        
        # Return the data as a JSON string
        result = {
            "file_path": file_path,
            "total_rows": total_rows,
            "columns": columns,
            "sample_data": sample_data,
            "missing_values": {str(k): int(v) for k, v in missing_values.items()}
        }
        print(f"Result: {result}")

        # Remove the ctx.send_text calls that aren't compatible with Temporal
        # Just return the result directly
        return json.dumps(result, indent=2)
    
    except Exception as e:
        error_msg = f"Error reading CSV file: {str(e)}"
        # Don't use send_text in Temporal workflows
        return json.dumps({"error": error_msg})

See ToolsWorkflow for the complete sample. The relevant worker registration.

3

Thanks @maxim.