Compare commits

..

2 Commits

Author SHA1 Message Date
github-actions[bot]
aaa478159d chore: update tool specifications 2026-02-20 10:03:20 +00:00
Greyson LaLonde
ddcfffe3ab refactor: simplify platform integration token resolution
Remove unused platform_context context manager and env var fallback
from context module. Token resolves from context var or env var via
default_factory on the tool field. Replace custom __init__ with
model_validator and use sanitize_tool_name.
2026-02-20 05:01:52 -05:00
9 changed files with 46 additions and 279 deletions

View File

@@ -6,8 +6,10 @@ from typing import Any
from crewai.tools import BaseTool
from crewai.utilities.pydantic_schema_utils import create_model_from_schema
from pydantic import Field, create_model
from crewai.utilities.string_utils import sanitize_tool_name
from pydantic import Field, create_model, model_validator
import requests
from typing_extensions import Self
from crewai_tools.tools.crewai_platform_tools.misc import (
get_platform_api_base_url,
@@ -20,34 +22,27 @@ class CrewAIPlatformActionTool(BaseTool):
action_schema: dict[str, Any] = Field(
default_factory=dict, description="The schema of the action"
)
integration_token: str | None = Field(
default_factory=get_platform_integration_token,
)
def __init__(
self,
description: str,
action_name: str,
action_schema: dict[str, Any],
):
parameters = action_schema.get("function", {}).get("parameters", {})
@model_validator(mode="after")
def _build_args_schema(self) -> Self:
parameters = self.action_schema.get("function", {}).get("parameters", {})
if parameters and parameters.get("properties"):
try:
if "title" not in parameters:
parameters = {**parameters, "title": f"{action_name}Schema"}
parameters = {**parameters, "title": f"{self.action_name}Schema"}
if "type" not in parameters:
parameters = {**parameters, "type": "object"}
args_schema = create_model_from_schema(parameters)
self.args_schema = create_model_from_schema(parameters)
except Exception:
args_schema = create_model(f"{action_name}Schema")
self.args_schema = create_model(f"{self.action_name}Schema")
else:
args_schema = create_model(f"{action_name}Schema")
super().__init__(
name=action_name.lower().replace(" ", "_"),
description=description,
args_schema=args_schema,
)
self.action_name = action_name
self.action_schema = action_schema
self.args_schema = create_model(f"{self.action_name}Schema")
if not self.name:
self.name = sanitize_tool_name(self.action_name)
return self
def _run(self, **kwargs: Any) -> str:
try:
@@ -58,9 +53,8 @@ class CrewAIPlatformActionTool(BaseTool):
api_url = (
f"{get_platform_api_base_url()}/actions/{self.action_name}/execute"
)
token = get_platform_integration_token()
headers = {
"Authorization": f"Bearer {token}",
"Authorization": f"Bearer {self.integration_token}",
"Content-Type": "application/json",
}
payload = {

View File

@@ -6,6 +6,7 @@ from types import TracebackType
from typing import Any
from crewai.tools import BaseTool
from crewai.utilities.string_utils import sanitize_tool_name
import requests
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import (
@@ -30,6 +31,7 @@ class CrewaiPlatformToolBuilder:
self._apps = apps
self._actions_schema: dict[str, dict[str, Any]] = {}
self._tools: list[BaseTool] | None = None
self._integration_token = get_platform_integration_token()
def tools(self) -> list[BaseTool]:
"""Fetch actions and return built tools."""
@@ -41,7 +43,7 @@ class CrewaiPlatformToolBuilder:
def _fetch_actions(self) -> None:
"""Fetch action schemas from the platform API."""
actions_url = f"{get_platform_api_base_url()}/actions"
headers = {"Authorization": f"Bearer {get_platform_integration_token()}"}
headers = {"Authorization": f"Bearer {self._integration_token}"}
try:
response = requests.get(
@@ -88,9 +90,11 @@ class CrewaiPlatformToolBuilder:
description = function_details.get("description", f"Execute {action_name}")
tool = CrewAIPlatformActionTool(
name=sanitize_tool_name(action_name),
description=description,
action_name=action_name,
action_schema=action_schema,
integration_token=self._integration_token,
)
tools.append(tool)

View File

@@ -9,17 +9,5 @@ def get_platform_api_base_url() -> str:
return f"{base_url}/crewai_plus/api/v1/integrations"
def get_platform_integration_token() -> str:
"""Get the platform integration token from the context.
Fallback to the environment variable if no token has been set in the context.
Raises:
ValueError: If no token has been set in the context.
"""
token = _get_context_token() or os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN")
if not token:
raise ValueError(
"No platform integration token found. "
"Set it via platform_integration_context() or set_platform_integration_token()."
)
return token
def get_platform_integration_token() -> str | None:
return _get_context_token() or os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN")

View File

@@ -1,56 +0,0 @@
"""Tests for platform tools misc functionality."""
import os
from unittest.mock import patch
import pytest
from crewai.context import platform_integration_context, set_platform_integration_token, reset_platform_integration_token
from crewai_tools.tools.crewai_platform_tools.misc import (
get_platform_integration_token,
)
class TestTokenRetrievalWithFallback:
"""Test token retrieval logic with environment fallback."""
@pytest.fixture
def clean_context(self):
token = set_platform_integration_token(None)
env_backup = os.environ.pop("CREWAI_PLATFORM_INTEGRATION_TOKEN", None)
yield
reset_platform_integration_token(token)
if env_backup is not None:
os.environ["CREWAI_PLATFORM_INTEGRATION_TOKEN"] = env_backup
else:
os.environ.pop("CREWAI_PLATFORM_INTEGRATION_TOKEN", None)
def test_context_token_takes_precedence(self, clean_context):
"""Test that context token takes precedence over environment variable."""
context_token = "context-token"
env_token = "env-token"
with patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": env_token}):
with platform_integration_context(context_token):
token = get_platform_integration_token()
assert token == context_token
def test_environment_fallback_when_no_context(self, clean_context):
"""Test fallback to environment variable when no context token."""
env_token = "env-fallback-token"
with patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": env_token}):
token = get_platform_integration_token()
assert token == env_token
@pytest.mark.parametrize("empty_value", ["", None])
def test_missing_token_raises_error(self, clean_context, empty_value):
"""Test that missing tokens raise appropriate errors."""
env_dict = {"CREWAI_PLATFORM_INTEGRATION_TOKEN": empty_value} if empty_value is not None else {}
with patch.dict(os.environ, env_dict, clear=True):
with pytest.raises(ValueError) as exc_info:
get_platform_integration_token()
assert "No platform integration token found" in str(exc_info.value)
assert "platform_integration_context()" in str(exc_info.value)

View File

@@ -27,9 +27,10 @@ class TestCrewAIPlatformActionToolVerify:
def create_test_tool(self):
return CrewAIPlatformActionTool(
name="test_action",
description="Test action tool",
action_name="test_action",
action_schema=self.action_schema
action_schema=self.action_schema,
)
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"}, clear=True)

View File

@@ -107,12 +107,10 @@ class TestCrewaiPlatformToolBuilder(unittest.TestCase):
)
def test_fetch_actions_no_token(self):
builder = CrewaiPlatformToolBuilder(apps=["github"])
with patch.dict("os.environ", {}, clear=True):
with self.assertRaises(ValueError) as context:
builder._fetch_actions()
assert "No platform integration token found" in str(context.exception)
builder = CrewaiPlatformToolBuilder(apps=["github"])
assert builder._integration_token is None
assert builder.tools() == []
@patch.dict("os.environ", {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"})
@patch(

View File

@@ -110,6 +110,5 @@ class TestCrewaiPlatformTools(unittest.TestCase):
def test_crewai_platform_tools_no_token(self):
with patch.dict("os.environ", {}, clear=True):
with self.assertRaises(ValueError) as context:
CrewaiPlatformTools(apps=["github"])
assert "No platform integration token found" in str(context.exception)
tools = CrewaiPlatformTools(apps=["github"])
assert tools == []

View File

@@ -1,7 +1,4 @@
from collections.abc import Generator
from contextlib import AbstractContextManager, contextmanager, nullcontext
import contextvars
from typing import Any
_platform_integration_token: contextvars.ContextVar[str | None] = (
@@ -9,51 +6,11 @@ _platform_integration_token: contextvars.ContextVar[str | None] = (
)
def set_platform_integration_token(integration_token: str) -> contextvars.Token[str | None]:
"""Set the platform integration token in the current context.
Args:
integration_token: The integration token to set.
"""
return _platform_integration_token.set(integration_token)
def reset_platform_integration_token(token: contextvars.Token[str | None]) -> None:
"""Reset the platform integration token to its previous value."""
_platform_integration_token.reset(token)
def get_platform_integration_token() -> str | None:
"""Get the platform integration token from the current context.
Returns:
The integration token if set, otherwise None.
"""
"""Get the platform integration token from the current context."""
return _platform_integration_token.get()
def platform_integration_context(integration_token: str | None) -> AbstractContextManager[None]:
"""Context manager to temporarily set the platform integration token.
Args:
integration_token: The integration token to set within the context.
If None or falsy, returns nullcontext (no-op).
Returns:
A context manager that either sets the token or does nothing.
"""
if not integration_token:
return nullcontext()
@contextmanager
def _token_context() -> Generator[None, Any, None]:
token = set_platform_integration_token(integration_token)
try:
yield
finally:
reset_platform_integration_token(token)
return _token_context()
_current_task_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
"current_task_id", default=None
)

View File

@@ -1,145 +1,27 @@
# ruff: noqa: S105
import os
from unittest.mock import patch
import pytest
from crewai.context import (
_platform_integration_token,
get_platform_integration_token,
platform_integration_context,
reset_platform_integration_token,
set_platform_integration_token,
)
@pytest.fixture
def clean_context():
"""Fixture to ensure clean context state for each test."""
_platform_integration_token.set(None)
yield
_platform_integration_token.set(None)
class TestPlatformIntegrationToken:
def setup_method(self):
_platform_integration_token.set(None)
def teardown_method(self):
_platform_integration_token.set(None)
class TestContextVariableCore:
"""Test core context variable functionality (set/get/reset)."""
def test_set_and_get_token(self, clean_context):
"""Test basic token setting and retrieval."""
test_token = "test-token-123"
def test_set_and_get(self):
assert get_platform_integration_token() is None
_platform_integration_token.set("test-token-123")
assert get_platform_integration_token() == "test-token-123"
def test_returns_none_when_not_set(self):
assert get_platform_integration_token() is None
context_token = set_platform_integration_token(test_token)
assert get_platform_integration_token() == test_token
assert context_token is not None
def test_reset_token_restores_previous_state(self, clean_context):
"""Test that reset properly restores previous context state."""
token1 = "token-1"
token2 = "token-2"
context_token1 = set_platform_integration_token(token1)
assert get_platform_integration_token() == token1
context_token2 = set_platform_integration_token(token2)
assert get_platform_integration_token() == token2
reset_platform_integration_token(context_token2)
assert get_platform_integration_token() == token1
reset_platform_integration_token(context_token1)
assert get_platform_integration_token() is None
def test_nested_token_management(self, clean_context):
"""Test proper token management with deeply nested contexts."""
tokens = ["token-1", "token-2", "token-3"]
context_tokens = []
for token in tokens:
context_tokens.append(set_platform_integration_token(token))
assert get_platform_integration_token() == token
for i in range(len(tokens) - 1, 0, -1):
reset_platform_integration_token(context_tokens[i])
assert get_platform_integration_token() == tokens[i - 1]
reset_platform_integration_token(context_tokens[0])
assert get_platform_integration_token() is None
@patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token"})
def test_context_module_ignores_environment_variables(self, clean_context):
"""Test that context module only returns context values, not env vars."""
# Context module should not read environment variables
assert get_platform_integration_token() is None
# Only context variable should be returned
set_platform_integration_token("context-token")
assert get_platform_integration_token() == "context-token"
class TestPlatformIntegrationContext:
"""Test platform integration context manager behavior."""
def test_basic_context_manager_usage(self, clean_context):
"""Test basic context manager functionality."""
test_token = "context-token"
assert get_platform_integration_token() is None
with platform_integration_context(test_token):
assert get_platform_integration_token() == test_token
assert get_platform_integration_token() is None
@pytest.mark.parametrize("falsy_value", [None, "", False, 0])
def test_falsy_values_return_nullcontext(self, clean_context, falsy_value):
"""Test that falsy values return nullcontext (no-op)."""
# Set initial token to verify nullcontext doesn't affect it
initial_token = "initial-token"
initial_context_token = set_platform_integration_token(initial_token)
try:
with platform_integration_context(falsy_value):
# Should preserve existing context (nullcontext behavior)
assert get_platform_integration_token() == initial_token
# Should still have initial token after nullcontext
assert get_platform_integration_token() == initial_token
finally:
reset_platform_integration_token(initial_context_token)
@pytest.mark.parametrize("truthy_value", ["token", "123", " ", "0"])
def test_truthy_values_create_context(self, clean_context, truthy_value):
"""Test that truthy values create proper context."""
with platform_integration_context(truthy_value):
assert get_platform_integration_token() == truthy_value
# Should be cleaned up
assert get_platform_integration_token() is None
def test_context_preserves_existing_token(self, clean_context):
"""Test that context manager preserves existing token when exiting."""
existing_token = "existing-token"
context_token = "context-token"
existing_context_token = set_platform_integration_token(existing_token)
try:
with platform_integration_context(context_token):
assert get_platform_integration_token() == context_token
assert get_platform_integration_token() == existing_token
finally:
reset_platform_integration_token(existing_context_token)
def test_context_manager_return_type(self, clean_context):
"""Test that context manager returns proper types for both cases."""
# Both should be usable as context managers
valid_ctx = platform_integration_context("token")
none_ctx = platform_integration_context(None)
assert hasattr(valid_ctx, '__enter__')
assert hasattr(valid_ctx, '__exit__')
assert hasattr(none_ctx, '__enter__')
assert hasattr(none_ctx, '__exit__')
def test_overwrite(self):
_platform_integration_token.set("first")
_platform_integration_token.set("second")
assert get_platform_integration_token() == "second"