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 44 additions and 303 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

@@ -1,5 +1,7 @@
import os
from crewai.context import get_platform_integration_token as _get_context_token
def get_platform_api_base_url() -> str:
"""Get the platform API base URL from environment or use default."""
@@ -7,11 +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 API base URL from environment or use default."""
token = os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN") or ""
if not token:
raise ValueError(
"No platform integration token found, please set the CREWAI_PLATFORM_INTEGRATION_TOKEN environment variable"
)
return token # TODO: Use context manager to get token
def get_platform_integration_token() -> str | None:
return _get_context_token() or os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN")

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

@@ -20117,18 +20117,6 @@
"humanized_name": "Web Automation Tool",
"init_params_schema": {
"$defs": {
"AvailableModel": {
"enum": [
"gpt-4o",
"gpt-4o-mini",
"claude-3-5-sonnet-latest",
"claude-3-7-sonnet-latest",
"computer-use-preview",
"gemini-2.0-flash"
],
"title": "AvailableModel",
"type": "string"
},
"EnvVar": {
"properties": {
"default": {
@@ -20206,17 +20194,6 @@
"default": null,
"title": "Model Api Key"
},
"model_name": {
"anyOf": [
{
"$ref": "#/$defs/AvailableModel"
},
{
"type": "null"
}
],
"default": "claude-3-7-sonnet-latest"
},
"project_id": {
"anyOf": [
{

View File

@@ -1,8 +1,4 @@
from collections.abc import Generator
from contextlib import contextmanager
import contextvars
import os
from typing import Any
_platform_integration_token: contextvars.ContextVar[str | None] = (
@@ -10,39 +6,9 @@ _platform_integration_token: contextvars.ContextVar[str | None] = (
)
def set_platform_integration_token(integration_token: str) -> None:
"""Set the platform integration token in the current context.
Args:
integration_token: The integration token to set.
"""
_platform_integration_token.set(integration_token)
def get_platform_integration_token() -> str | None:
"""Get the platform integration token from the current context or environment.
Returns:
The integration token if set, otherwise None.
"""
token = _platform_integration_token.get()
if token is None:
token = os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN")
return token
@contextmanager
def platform_context(integration_token: str) -> Generator[None, Any, None]:
"""Context manager to temporarily set the platform integration token.
Args:
integration_token: The integration token to set within the context.
"""
token = _platform_integration_token.set(integration_token)
try:
yield
finally:
_platform_integration_token.reset(token)
"""Get the platform integration token from the current context."""
return _platform_integration_token.get()
_current_task_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(

View File

@@ -1,14 +1,8 @@
# ruff: noqa: S105
import os
from unittest.mock import patch
import pytest
from crewai.context import (
_platform_integration_token,
get_platform_integration_token,
platform_context,
set_platform_integration_token,
)
@@ -19,203 +13,15 @@ class TestPlatformIntegrationToken:
def teardown_method(self):
_platform_integration_token.set(None)
@patch.dict(os.environ, {}, clear=True)
def test_set_platform_integration_token(self):
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
set_platform_integration_token(test_token)
assert get_platform_integration_token() == test_token
def test_get_platform_integration_token_from_context_var(self):
test_token = "context-var-token"
_platform_integration_token.set(test_token)
assert get_platform_integration_token() == test_token
@patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token-456"})
def test_get_platform_integration_token_from_env_var(self):
assert _platform_integration_token.get() is None
assert get_platform_integration_token() == "env-token-456"
@patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token"})
def test_context_var_takes_precedence_over_env_var(self):
context_token = "context-token"
set_platform_integration_token(context_token)
assert get_platform_integration_token() == context_token
@patch.dict(os.environ, {}, clear=True)
def test_get_platform_integration_token_returns_none_when_not_set(self):
assert _platform_integration_token.get() is None
assert get_platform_integration_token() is None
@patch.dict(os.environ, {}, clear=True)
def test_platform_context_manager_basic_usage(self):
test_token = "context-manager-token"
assert get_platform_integration_token() is None
with platform_context(test_token):
assert get_platform_integration_token() == test_token
assert get_platform_integration_token() is None
@patch.dict(os.environ, {}, clear=True)
def test_platform_context_manager_nested_contexts(self):
"""Test nested platform_context context managers."""
outer_token = "outer-token"
inner_token = "inner-token"
assert get_platform_integration_token() is None
with platform_context(outer_token):
assert get_platform_integration_token() == outer_token
with platform_context(inner_token):
assert get_platform_integration_token() == inner_token
assert get_platform_integration_token() == outer_token
assert get_platform_integration_token() is None
def test_platform_context_manager_preserves_existing_token(self):
"""Test that platform_context preserves existing token when exiting."""
initial_token = "initial-token"
context_token = "context-token"
set_platform_integration_token(initial_token)
assert get_platform_integration_token() == initial_token
with platform_context(context_token):
assert get_platform_integration_token() == context_token
assert get_platform_integration_token() == initial_token
def test_platform_context_manager_exception_handling(self):
"""Test that platform_context properly resets token even when exception occurs."""
initial_token = "initial-token"
context_token = "context-token"
set_platform_integration_token(initial_token)
with pytest.raises(ValueError):
with platform_context(context_token):
assert get_platform_integration_token() == context_token
raise ValueError("Test exception")
assert get_platform_integration_token() == initial_token
@patch.dict(os.environ, {}, clear=True)
def test_platform_context_manager_with_none_initial_state(self):
"""Test platform_context when initial state is None."""
context_token = "context-token"
assert get_platform_integration_token() is None
with pytest.raises(RuntimeError):
with platform_context(context_token):
assert get_platform_integration_token() == context_token
raise RuntimeError("Test exception")
assert get_platform_integration_token() is None
@patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-backup"})
def test_platform_context_with_env_fallback(self):
"""Test platform_context interaction with environment variable fallback."""
context_token = "context-token"
assert get_platform_integration_token() == "env-backup"
with platform_context(context_token):
assert get_platform_integration_token() == context_token
assert get_platform_integration_token() == "env-backup"
@patch.dict(os.environ, {}, clear=True)
def test_multiple_sequential_context_managers(self):
"""Test multiple sequential uses of platform_context."""
token1 = "token-1"
token2 = "token-2"
token3 = "token-3"
with platform_context(token1):
assert get_platform_integration_token() == token1
assert get_platform_integration_token() is None
with platform_context(token2):
assert get_platform_integration_token() == token2
assert get_platform_integration_token() is None
with platform_context(token3):
assert get_platform_integration_token() == token3
assert get_platform_integration_token() is None
def test_empty_string_token(self):
empty_token = ""
set_platform_integration_token(empty_token)
assert get_platform_integration_token() == ""
with platform_context(empty_token):
assert get_platform_integration_token() == ""
def test_special_characters_in_token(self):
special_token = "token-with-!@#$%^&*()_+-={}[]|\\:;\"'<>?,./"
set_platform_integration_token(special_token)
assert get_platform_integration_token() == special_token
with platform_context(special_token):
assert get_platform_integration_token() == special_token
def test_very_long_token(self):
long_token = "a" * 10000
set_platform_integration_token(long_token)
assert get_platform_integration_token() == long_token
with platform_context(long_token):
assert get_platform_integration_token() == long_token
@patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": ""})
def test_empty_env_var(self):
assert _platform_integration_token.get() is None
assert get_platform_integration_token() == ""
@patch("crewai.context.os.getenv")
def test_env_var_access_error_handling(self, mock_getenv):
mock_getenv.side_effect = OSError("Environment access error")
with pytest.raises(OSError):
get_platform_integration_token()
@patch.dict(os.environ, {}, clear=True)
def test_context_var_isolation_between_tests(self):
"""Test that context variable changes don't leak between test methods."""
test_token = "isolation-test-token"
assert get_platform_integration_token() is None
set_platform_integration_token(test_token)
assert get_platform_integration_token() == test_token
def test_context_manager_return_value(self):
"""Test that platform_context can be used in with statement with return value."""
test_token = "return-value-token"
with platform_context(test_token):
assert get_platform_integration_token() == test_token
with platform_context(test_token) as ctx:
assert ctx is None
assert get_platform_integration_token() == test_token
def test_overwrite(self):
_platform_integration_token.set("first")
_platform_integration_token.set("second")
assert get_platform_integration_token() == "second"