Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 145 additions & 0 deletions onshape_mcp/api/featurestudio.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
"""Feature Studio API operations for Onshape.

Manages Feature Studio elements — create, read, and update FeatureScript code.
Used to deploy custom batch builder features that reduce API call count.
"""

from typing import Any, Dict, Optional

from .client import OnshapeClient


class FeatureStudioManager:
"""Manager for Feature Studio API operations."""

def __init__(self, client: OnshapeClient):
self.client = client

async def create(
self,
document_id: str,
workspace_id: str,
name: str = "MCP Builders",
) -> Dict[str, Any]:
"""Create a new Feature Studio in a document.

Args:
document_id: Document ID
workspace_id: Workspace ID
name: Name for the Feature Studio tab

Returns:
API response with element ID
"""
path = f"/api/v9/featurestudios/d/{document_id}/w/{workspace_id}"
data = {"name": name}
return await self.client.post(path, data=data)

async def get_contents(
self,
document_id: str,
workspace_id: str,
element_id: str,
) -> Dict[str, Any]:
"""Get the FeatureScript source code from a Feature Studio.

Args:
document_id: Document ID
workspace_id: Workspace ID
element_id: Feature Studio element ID

Returns:
Response containing 'contents' field with FeatureScript source
"""
path = f"/api/v9/featurestudios/d/{document_id}/w/{workspace_id}/e/{element_id}"
return await self.client.get(path)

async def update_contents(
self,
document_id: str,
workspace_id: str,
element_id: str,
contents: str,
source_microversion: Optional[str] = None,
) -> Dict[str, Any]:
"""Update the FeatureScript source code in a Feature Studio.

Args:
document_id: Document ID
workspace_id: Workspace ID
element_id: Feature Studio element ID
contents: FeatureScript source code
source_microversion: Optional microversion for conflict detection

Returns:
API response
"""
path = f"/api/v9/featurestudios/d/{document_id}/w/{workspace_id}/e/{element_id}"
data: Dict[str, Any] = {
"contents": contents,
"serializationVersion": "1.2.16",
"rejectMicroversionSkew": False,
}
if source_microversion:
data["sourceMicroversion"] = source_microversion
return await self.client.post(path, data=data)

async def get_specs(
self,
document_id: str,
workspace_id: str,
element_id: str,
) -> Dict[str, Any]:
"""Get the feature specs (parameter definitions) from a Feature Studio.

Returns the parameter schema for each custom feature defined in the
Feature Studio. Used to understand how to invoke custom features via
the addFeature API.

Args:
document_id: Document ID
workspace_id: Workspace ID
element_id: Feature Studio element ID

Returns:
Feature specs with parameter definitions
"""
path = (
f"/api/v9/featurestudios/d/{document_id}/w/{workspace_id}/e/{element_id}/featurespecs"
)
return await self.client.get(path)

async def deploy_builders(
self,
document_id: str,
workspace_id: str,
name: str = "MCP Builders",
) -> Dict[str, str]:
"""Create a Feature Studio and populate it with MCP builder features.

This is the main entry point for setting up batch operations in a
document. Creates the Feature Studio and writes the builder
FeatureScript code. Only needs to be called once per document.

Args:
document_id: Document ID
workspace_id: Workspace ID
name: Name for the Feature Studio

Returns:
Dict with 'elementId' of the created Feature Studio
"""
import importlib.resources

# Create the Feature Studio
result = await self.create(document_id, workspace_id, name)
element_id = result.get("id", "")

# Read the builder FeatureScript source
fs_source = importlib.resources.files("onshape_mcp.featurescript").joinpath("builders.fs")
contents = fs_source.read_text()

# Upload the source code
await self.update_contents(document_id, workspace_id, element_id, contents)

return {"elementId": element_id}
114 changes: 94 additions & 20 deletions onshape_mcp/api/variables.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,45 @@
"""Variable table management for Onshape Part Studios."""
"""Variable table management for Onshape Variable Studios."""

from typing import Any, Dict, List, Optional
from pydantic import BaseModel
from .client import OnshapeClient


# Common FeatureScript VariableType values
VARIABLE_TYPE_MAP = {
"in": "LENGTH",
"mm": "LENGTH",
"cm": "LENGTH",
"m": "LENGTH",
"ft": "LENGTH",
"deg": "ANGLE",
"rad": "ANGLE",
}


def _infer_variable_type(expression: str) -> str:
"""Infer the FeatureScript VariableType from an expression.

Args:
expression: Variable expression (e.g., '0.75 in', '45 deg', '3.14')

Returns:
VariableType string (LENGTH, ANGLE, or ANY)
"""
expr = expression.strip().lower()
for suffix, var_type in VARIABLE_TYPE_MAP.items():
if expr.endswith(suffix):
return var_type
return "ANY"


class Variable(BaseModel):
"""Represents a variable in an Onshape variable table."""

name: str
expression: str
description: Optional[str] = None
type: Optional[str] = None


class VariableManager:
Expand All @@ -24,31 +53,51 @@ def __init__(self, client: OnshapeClient):
"""
self.client = client

async def create_variable_studio(
self, document_id: str, workspace_id: str, name: str
) -> Dict[str, Any]:
"""Create a new Variable Studio in a document.

Args:
document_id: Document ID
workspace_id: Workspace ID
name: Name for the new Variable Studio

Returns:
API response with new Variable Studio info
"""
path = f"/api/variables/d/{document_id}/w/{workspace_id}/variablestudio"
data = {"name": name}
return await self.client.post(path, data=data)

async def get_variables(
self, document_id: str, workspace_id: str, element_id: str
) -> List[Variable]:
"""Get all variables from a Part Studio.
"""Get all variables from a Variable Studio.

Args:
document_id: Document ID
workspace_id: Workspace ID
element_id: Part Studio element ID
element_id: Variable Studio element ID

Returns:
List of variables
"""
path = f"/api/v6/variables/d/{document_id}/w/{workspace_id}/e/{element_id}/variables"
path = f"/api/variables/d/{document_id}/w/{workspace_id}/e/{element_id}/variables"
response = await self.client.get(path)

# Response is [{variableStudioReference, variables: [...]}, ...]
variables = []
for var_data in response:
variables.append(
Variable(
name=var_data.get("name", ""),
expression=var_data.get("expression", ""),
description=var_data.get("description"),
for group in response:
for var_data in group.get("variables", []):
variables.append(
Variable(
name=var_data.get("name", ""),
expression=var_data.get("expression", ""),
description=var_data.get("description"),
type=var_data.get("type"),
)
)
)

return variables

Expand All @@ -61,27 +110,52 @@ async def set_variable(
expression: str,
description: Optional[str] = None,
) -> Dict[str, Any]:
"""Set or update a variable in a Part Studio.
"""Set or update a variable in a Variable Studio.

Preserves existing variables by reading them first, then sending
the full list with the new/updated variable.

Args:
document_id: Document ID
workspace_id: Workspace ID
element_id: Part Studio element ID
element_id: Variable Studio element ID
name: Variable name
expression: Variable expression (e.g., "0.75 in")
description: Optional variable description

Returns:
API response
"""
path = f"/api/v6/variables/d/{document_id}/w/{workspace_id}/e/{element_id}/variables"

data = [{"name": name, "expression": expression}]

path = f"/api/variables/d/{document_id}/w/{workspace_id}/e/{element_id}/variables"

# GET existing variables so we don't overwrite them
existing = await self.client.get(path)
variables: List[Dict[str, Any]] = []
for group in existing:
for var_data in group.get("variables", []):
variables.append({
"name": var_data["name"],
"expression": var_data["expression"],
"type": var_data.get("type", "ANY"),
**({"description": var_data["description"]} if var_data.get("description") else {}),
})

# Update existing or append new
var_type = _infer_variable_type(expression)
new_var: Dict[str, Any] = {"name": name, "expression": expression, "type": var_type}
if description:
data[0]["description"] = description

return await self.client.post(path, data=data)
new_var["description"] = description

found = False
for i, v in enumerate(variables):
if v["name"] == name:
variables[i] = new_var
found = True
break
if not found:
variables.append(new_var)

return await self.client.post(path, data=variables)

async def get_configuration_definition(
self, document_id: str, workspace_id: str, element_id: str
Expand Down
Loading
Loading