Limit tools available to a user

Author: rafaljanicki

README.md

@@ -133,7 +133,6 @@ The following tools are exposed, derived from the FastAPI application's endpoint
 - **Description**: Starts a new sandbox instance for a specified use case.
 - **Input**:
     - `usecase` (str): The name of the use case for the sandbox (e.g., 'movies', 'blank').
-    - `cease_emails` (Optional[bool], default: `False`): If true, no emails will be sent regarding this sandbox.
 - **Output**: `Dict` (JSON object representing the newly started sandbox)
 
 ---
@@ -150,7 +149,6 @@ The following tools are exposed, derived from the FastAPI application's endpoint
 - **Description**: Extends the lifetime of a sandbox or all sandboxes for the user.
 - **Input**:
     - `sandbox_hash_key` (Optional[str]): Specific sandbox to extend. If None, all user's sandboxes are extended.
-    - `profile_data` (Optional[Dict[str, Any]]): User profile information.
 - **Output**: `Dict` (JSON object with status of the extension)
 
 ---
@@ -164,23 +162,6 @@ The following tools are exposed, derived from the FastAPI application's endpoint
 
 ---
 
-### `invite_sandbox_collaborator`
-- **Description**: Invites a collaborator to a specific sandbox.
-- **Input**:
-    - `sandbox_hash_key` (str): The unique hash key identifying the sandbox to share.
-    - `email` (str): Email address of the user to invite.
-    - `message` (str): A personal message to include in the invitation.
-- **Output**: `Dict` (JSON object with the status of the invitation)
-
----
-
-### `get_user_information`
-- **Description**: Retrieves user information for the authenticated user (from Sandbox API if needed, or use token data).
-- **Input**: None
-- **Output**: `Dict` (JSON object containing user information)
-
----
-
 ### `request_sandbox_backup`
 - **Description**: Requests a backup for a specific sandbox.
 - **Input**:

src/sandbox_api_mcp_server/sandbox/models.py

@@ -1,10 +1,9 @@
-from typing import Annotated, Optional, Dict, Any
+from typing import Annotated, Optional, Any
 from pydantic import BaseModel, Field
 
 
 class StartSandboxBody(BaseModel):
     usecase: Annotated[str, Field(description="The name of the use case for the sandbox (e.g., 'movies', 'blank').")]
-    cease_emails: Annotated[Optional[bool], Field(description="If true, no emails will be sent regarding this sandbox.")] = False
 
 
 class StopSandboxBody(BaseModel):
@@ -13,7 +12,6 @@ class StopSandboxBody(BaseModel):
 
 class ExtendSandboxBody(BaseModel):
     sandbox_hash_key: Annotated[Optional[str], Field(description="Specific sandbox to extend. If None, all user's sandboxes are extended.")] = None
-    profile_data: Annotated[Optional[Dict[str, Any]], Field(description="User profile information.")] = None
 
 
 class InviteCollaboratorsBody(BaseModel):

src/sandbox_api_mcp_server/sandbox/routes.py

@@ -4,7 +4,7 @@
 from fastapi import Query, Path, status
 from fastapi.responses import PlainTextResponse
 
-from .models import StartSandboxBody, StopSandboxBody, ExtendSandboxBody, InviteCollaboratorsBody, AuraUploadBody, BackupDownloadUrlBody, FastApiReadCypherQueryBody, FastApiWriteCypherQueryBody
+from .models import StartSandboxBody, StopSandboxBody, ExtendSandboxBody, AuraUploadBody, BackupDownloadUrlBody, FastApiReadCypherQueryBody, FastApiWriteCypherQueryBody
 from helpers import get_logger
 from .service import call_sandbox_api, SandboxApiClient, get_sandbox_client
 
@@ -35,7 +35,7 @@ async def start_sandbox(
     ):
         """Starts a new sandbox instance for a specified use case."""
         try:
-            return await call_sandbox_api("start_sandbox", client, usecase=body.usecase, cease_emails=body.cease_emails)
+            return await call_sandbox_api("start_sandbox", client, usecase=body.usecase)
         except Exception as e:
             logger.error(f"Error starting sandbox: {e}")
             raise e
@@ -59,7 +59,7 @@ async def extend_sandbox(
     ):
         """Extends the lifetime of a sandbox or all sandboxes for the user."""
         try:
-            return await call_sandbox_api("extend_sandbox", client, sandbox_hash_key=body.sandbox_hash_key, profile_data=body.profile_data)
+            return await call_sandbox_api("extend_sandbox", client, sandbox_hash_key=body.sandbox_hash_key)
         except Exception as e:
             logger.error(f"Error extending sandbox: {e}")
             raise e
@@ -78,31 +78,6 @@ async def get_sandbox_details(
             logger.error(f"Error getting sandbox details: {e}")
             raise e
 
-    @router.post("/invite-collaborator", operation_id="invite_sandbox_collaborator", tags=["Sandbox"], response_model=Dict)
-    async def invite_collaborator(
-        body: InviteCollaboratorsBody,
-        client: Annotated[SandboxApiClient, Depends(get_sandbox_client)],
-    ):
-        """Invites a collaborator to a specific sandbox."""
-        try:
-            return await call_sandbox_api("invite_collaborator", client, sandbox_hash_key=body.sandbox_hash_key, email=body.email, message=body.message)
-        except Exception as e:
-            logger.error(f"Error inviting collaborator: {e}")
-            raise e
-
-    @router.get("/user-info", operation_id="get_user_information", tags=["User"], response_model=Dict)
-    async def get_user_info_ep(
-        client: Annotated[SandboxApiClient, Depends(get_sandbox_client)],
-    ):
-        """Retrieves user information for the authenticated user (from Sandbox API if needed, or use token data)."""
-        # This could either return user info from the token (user param) or call the Sandbox API
-        # return user # Example: return Auth0 user profile
-        try:
-            return await call_sandbox_api("get_user_info", client)
-        except Exception as e:
-            logger.error(f"Error getting user info: {e}")
-            raise e
-
     # --- Backup Related Endpoints ---
     @router.post("/request-backup/{sandbox_hash_key}", operation_id="request_sandbox_backup", tags=["Backup"], response_model=Dict)
     async def request_backup_ep(
@@ -161,7 +136,14 @@ async def upload_to_aura_ep(
     ):
         """Uploads a sandbox backup to an Aura instance."""
         try:
-            return await call_sandbox_api("upload_to_aura", client, sandbox_hash_key=body.sandbox_hash_key, aura_uri=body.aura_uri, aura_password=body.aura_password, aura_username=body.aura_username)
+            return await call_sandbox_api(
+                "upload_to_aura",
+                client,
+                sandbox_hash_key=body.sandbox_hash_key,
+                aura_uri=body.aura_uri,
+                aura_password=body.aura_password,
+                aura_username=body.aura_username,
+            )
         except Exception as e:
             logger.error(f"Error uploading to Aura: {e}")
             raise e

src/sandbox_api_mcp_server/sandbox/service.py

@@ -79,29 +79,23 @@ async def list_sandboxes_for_user(self, timezone: Optional[str] = None) -> Dict[
             "sandboxes": await self._request("GET", "/SandboxGetRunningInstancesForUser", params=params),
         }
 
-    async def start_sandbox(self, usecase: str, cease_emails: Optional[bool] = None, draft_only: Optional[bool] = None, auth_token: Optional[str] = None) -> Dict[str, Any]:
+    async def start_sandbox(self, usecase: str) -> Dict[str, Any]:
         """
         Creates and deploys a new sandbox instance or returns an existing one if duplicates are not allowed and one exists.
         Corresponds to POST /SandboxRunInstance in swagger.
 
         Args:
-            usecase (str): The name of the use case for the sandbox.
-            cease_emails (Optional[bool]): If true, no emails will be sent regarding this sandbox. (default: false)
-            draft_only (Optional[bool]): If true, only a draft record is created, and the instance is not deployed. (default: false)
-            auth_token (Optional[str]): Optional authentication token for certain flows.
+            usecase (str): The name of the use case for the sandbox. Possible values are:
+                blank-sandbox,bloom,citations,contact-tracing,cybersecurity,entity-resolution,fincen,
+                fraud-detection,graph-data-science,graph-data-science-blank-sandbox,healthcare-analytics,
+                icij-offshoreleaks,icij-paradise-papers,legis-graph,movies,network-management,
+                openstreetmap,pole,recommendations,twitch,twitter-trolls,wwc2019,yelp,twitter-v2
 
         Returns:
             Dict[str, Any]: Sandbox instance details or draft confirmation.
                             See #/components/schemas/RunInstanceResponse in swagger.
         """
-        json_data: Dict[str, Any] = {"usecase": usecase}
-        if cease_emails is not None:
-            json_data["cease_emails"] = cease_emails
-        if draft_only is not None:
-            json_data["draft_only"] = draft_only
-        if auth_token is not None:
-            json_data["auth_token"] = auth_token
-        return await self._request("POST", "/SandboxRunInstance", json_data=json_data)
+        return await self._request("POST", "/SandboxRunInstance", json_data={"usecase": usecase})
 
     async def stop_sandbox(self, sandbox_hash_key: str) -> Optional[Dict[str, Any]]:
         """
@@ -118,24 +112,21 @@ async def stop_sandbox(self, sandbox_hash_key: str) -> Optional[Dict[str, Any]]:
         json_data = {"sandboxHashKey": sandbox_hash_key}
         return await self._request("POST", "/SandboxStopInstance", json_data=json_data)
 
-    async def extend_sandbox(self, sandbox_hash_key: Optional[str] = None, profile_data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    async def extend_sandbox(self, sandbox_hash_key: Optional[str] = None) -> Dict[str, Any]:
         """
         Extends the lifetime of a user's sandbox(es). User profile details can be submitted with this request.
         Corresponds to POST /SandboxExtend in swagger.
 
         Args:
             sandbox_hash_key (Optional[str]): Specific sandbox to extend. If not provided, all user's sandboxes are extended.
-            profile_data (Optional[Dict[str, Any]]): User profile information to submit.
-                Expected keys can include 'email', 'company', 'country', 'industry', 'telephone', 'jobrole', 'whyneo4j'.
-                See #/components/schemas/ExtendRequest in swagger.
 
         Returns:
             Dict[str, Any]: Sandbox lifetime extension status. See #/components/schemas/ExtendResponse.
         """
-        json_data = profile_data if profile_data is not None else {}
+        json_data: Dict[str, Any] = {}
         if sandbox_hash_key:
             json_data["sandboxHashKey"] = sandbox_hash_key
-        return await self._request("POST", "/SandboxExtend", json_data=json_data if json_data else {})
+        return await self._request("POST", "/SandboxExtend", json_data=json_data)
 
     async def get_sandbox_details(self, sandbox_hash_key: str, verify_connect: Optional[bool] = False) -> Dict[str, Any]:
         """
@@ -155,26 +146,6 @@ async def get_sandbox_details(self, sandbox_hash_key: str, verify_connect: Optio
             params["verifyConnect"] = verify_connect
         return await self._request("GET", "/SandboxAuthdGetInstanceByHashKey", params=params)
 
-    async def invite_collaborator(self, sandbox_hash_key: str, email: str, message: str) -> Dict[str, Any]:
-        """
-        Invites another user to share one of the authenticated user's sandboxes.
-        Corresponds to POST /SandboxShare in swagger.
-
-        Args:
-            sandbox_hash_key (str): The unique hash key identifying the sandbox to share.
-            email (str): Email address of the user to invite.
-            message (str): A personal message to include in the invitation.
-
-        Returns:
-            Dict[str, Any]: Invitation creation status. See #/components/schemas/ShareResponse.
-        """
-        json_data = {
-            "sandboxHashKey": sandbox_hash_key,
-            "email": email,
-            "message": message
-        }
-        return await self._request("POST", "/SandboxShare", json_data=json_data)
-
     async def request_backup(self, sandbox_hash_key: str) -> Dict[str, Any]:
         """
         Initiates a backup process for a specific sandbox.
@@ -272,22 +243,22 @@ async def get_aura_upload_result(self, result_id: str) -> Dict[str, Any]:
         endpoint = f"/SandboxAuraUpload/result/{result_id}"
         return await self._request("GET", endpoint)
 
-    async def get_user_info(self) -> Dict[str, Any]:
-        """
-        Retrieves user information for the authenticated user.
-        Corresponds to GET /SandboxGetUserInfo in swagger.
-
-        Returns:
-            Dict[str, Any]: User information. See #/components/schemas/UserInfoResponse.
-        """
-        return await self._request("GET", "/SandboxGetUserInfo")
-
     async def get_schema(self, hash_key: str) -> FastApiReadCypherQueryResponse:
         """
         Retrieves the schema of the Neo4j database.
         Corresponds to POST /SandboxQuery in swagger.
         """
-        return await self.read_query(hash_key, "call apoc.meta.data() yield label, property, type, other, unique, index, elementType where elementType = 'node' and not label starts with '_' with label, collect(case when type <> 'RELATIONSHIP' then [property, type + case when unique then ' unique' else '' end + case when index then ' indexed' else '' end] end) as attributes, collect(case when type = 'RELATIONSHIP' then [property, head(other)] end) as relationships return label, apoc.map.fromPairs(attributes) as attributes, apoc.map.fromPairs(relationships) as relationships")
+        schema_query = (
+            "call apoc.meta.data() yield label, property, type, other, unique, index, elementType "
+            "where elementType = 'node' and not label starts with '_' "
+            "with label, collect(case when type <> 'RELATIONSHIP' "
+            "then [property, type + case when unique then ' unique' else '' end + "
+            "case when index then ' indexed' else '' end] end) as attributes, "
+            "collect(case when type = 'RELATIONSHIP' then [property, head(other)] end) as relationships "
+            "return label, apoc.map.fromPairs(attributes) as attributes, "
+            "apoc.map.fromPairs(relationships) as relationships"
+        )
+        return await self.read_query(hash_key, schema_query)
 
     async def read_query(self, hash_key: str, query: str, params: Optional[Dict[str, Any]] = None) -> FastApiReadCypherQueryResponse:
         """