Python: .NET/Python: Validate CrewAI task identifiers (#14551)

### Motivation and Context

<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->

Align CrewAI task identifier handling across the .NET and Python clients
and fail early for unsupported identifier formats.

### Description

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

- Validate CrewAI status task identifiers consistently in both
implementations.
- Add focused unit coverage for accepted and rejected identifier
formats.
- Preserve valid identifiers unchanged in outgoing status requests.

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [x] The code builds clean without any errors or warnings
- [ ] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [ ] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄

---------

Co-authored-by: Copilot App <[email protected]>
This commit is contained in:
SergeyMenshykh
2026-10-06 11:39:59 +00:00
committed by GitHub
co-authored by Copilot App
parent 71b2193130
commit 141359393d
4 changed files with 160 additions and 1 deletions
@@ -1,5 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
import re
from typing import Any
import aiohttp
@@ -89,6 +90,9 @@ class CrewAIEnterpriseClient:
Returns:
CrewAIStatusResponse: The status response of the task.
"""
if not isinstance(task_id, str) or re.fullmatch(r"[A-Za-z0-9_-]+", task_id) is None:
raise ValueError("Task ID must contain only ASCII letters, digits, underscores, or hyphens.")
async with (
self.session.get(f"{self.endpoint}/status/{task_id}", headers=self.request_header) as response, # type: ignore
):
@@ -0,0 +1,72 @@
# Copyright (c) Microsoft. All rights reserved.
from unittest.mock import MagicMock
import aiohttp
import pytest
from yarl import URL
from semantic_kernel.core_plugins.crew_ai.crew_ai_enterprise_client import CrewAIEnterpriseClient
from semantic_kernel.core_plugins.crew_ai.crew_ai_models import CrewAIEnterpriseKickoffState
@pytest.fixture
def crew_ai_client():
client = CrewAIEnterpriseClient(endpoint="https://test.com", auth_token="FakeToken")
client.session = MagicMock(spec=aiohttp.ClientSession)
return client
@pytest.mark.parametrize(
"task_id",
["12345", "AZaz09_-", "640f9a3a-2734-4ba8-8b8a-2f796f419d03"],
)
async def test_get_status_preserves_valid_task_id(crew_ai_client, task_id):
response = MagicMock(spec=aiohttp.ClientResponse)
response.text.return_value = '{"state":"SUCCESS","result":"The Result"}'
crew_ai_client.session.get.return_value.__aenter__.return_value = response
result = await crew_ai_client.get_status(task_id)
assert result.state == CrewAIEnterpriseKickoffState.Success
assert result.result == "The Result"
crew_ai_client.session.get.assert_called_once_with(
f"https://test.com/status/{task_id}",
headers=crew_ai_client.request_header,
)
request_url = crew_ai_client.session.get.call_args.args[0]
assert URL(request_url).raw_path == f"/status/{task_id}"
response.raise_for_status.assert_called_once()
@pytest.mark.parametrize(
"task_id",
[
None,
123,
[],
"",
" ",
"task.id",
"task/id",
r"task\id",
"task%id",
"task?id",
"task#id",
"task id",
"task\n",
"task\r\n",
"task\t",
"task\x00",
"t\u00e1sk",
"\uff11\uff12\uff13",
],
)
async def test_get_status_rejects_invalid_task_id_without_request(crew_ai_client, task_id):
with pytest.raises(
ValueError,
match="Task ID must contain only ASCII letters, digits, underscores, or hyphens",
):
await crew_ai_client.get_status(task_id)
crew_ai_client.session.get.assert_not_called()