spix_call_create

Instructions

Implementation Reference

• src/spix_mcp/tools.py:94-203 (handler)

  The `create_tool_handler` function is the main handler that executes MCP tool calls. It resolves the tool name 'spix_call_create' to the 'call.create' command schema, validates session scope, builds the endpoint URL, dispatches to the backend API (POST /calls), and returns the response.

  async def create_tool_handler(
      session: McpSessionContext,
      tool_name: str,
      arguments: dict,
  ) -> list:
      """Execute an MCP tool call by dispatching to the backend API.
  
      This function:
      1. Resolves the tool name to a command schema
      2. Validates session scope (playbook access, channel access)
      3. Builds the API request
      4. Dispatches to the backend
      5. Returns the response as MCP TextContent
  
      Args:
          session: The MCP session context for scope validation.
          tool_name: The MCP tool name (e.g., "spix_playbook_create").
          arguments: The tool arguments from the MCP client.
  
      Returns:
          List containing a single TextContent with the JSON response.
      """
      # Import here to avoid circular imports and handle missing mcp package
      try:
          from mcp.types import TextContent
      except ImportError:
          # Fallback for when mcp is not installed
          class TextContent:  # type: ignore[no-redef]
              def __init__(self, type: str, text: str) -> None:
                  self.type = type
                  self.text = text
  
      # Resolve tool name to schema
      schema = get_schema_by_tool_name(tool_name)
      if not schema:
          return [
              TextContent(
                  type="text",
                  text=orjson.dumps(
                      {"ok": False, "error": {"code": "unknown_tool", "message": f"Unknown tool: {tool_name}"}}
                  ).decode(),
              )
          ]
  
      # Validate tool access (not disabled)
      try:
          session.validate_tool_access(schema.path)
      except Exception as e:
          from spix_mcp.session import McpScopeError
  
          if isinstance(e, McpScopeError):
              return [TextContent(type="text", text=orjson.dumps({"ok": False, "error": e.to_dict()}).decode())]
          raise
  
      # Validate channel access if applicable
      channel = infer_channel_from_tool(schema.path)
      if channel:
          try:
              session.validate_channel_access(channel)
          except Exception as e:
              from spix_mcp.session import McpScopeError
  
              if isinstance(e, McpScopeError):
                  return [TextContent(type="text", text=orjson.dumps({"ok": False, "error": e.to_dict()}).decode())]
              raise
  
      # Handle playbook_id: validate and apply default
      playbook_id = arguments.get("playbook_id")
      try:
          effective_playbook = session.validate_playbook_access(playbook_id)
          if effective_playbook and not playbook_id:
              # Apply default playbook
              arguments["playbook_id"] = effective_playbook
      except Exception as e:
          from spix_mcp.session import McpScopeError
  
          if isinstance(e, McpScopeError):
              return [TextContent(type="text", text=orjson.dumps({"ok": False, "error": e.to_dict()}).decode())]
          raise
  
      # Build endpoint URL with path parameters
      endpoint, remaining_args = build_endpoint_url(schema, arguments)
  
      # Dispatch to backend API
      client = session.client
      method = schema.http_method.lower()
  
      if method == "get":
          response = await asyncio.to_thread(client.get, endpoint, params=remaining_args if remaining_args else None)
      elif method == "post":
          response = await asyncio.to_thread(client.post, endpoint, json=remaining_args if remaining_args else None)
      elif method == "patch":
          response = await asyncio.to_thread(client.patch, endpoint, json=remaining_args if remaining_args else None)
      elif method == "delete":
          response = await asyncio.to_thread(client.delete, endpoint, params=remaining_args if remaining_args else None)
      else:
          response = await asyncio.to_thread(client.get, endpoint)
  
      # Build response envelope
      envelope: dict = {"ok": response.ok, "meta": response.meta}
      if response.ok:
          envelope["data"] = response.data
          if response.pagination:
              envelope["pagination"] = response.pagination
          if response.warnings:
              envelope["warnings"] = response.warnings
      else:
          envelope["error"] = response.error
  
      return [TextContent(type="text", text=orjson.dumps(envelope).decode())]

• src/spix_mcp/server.py:84-97 (registration)

  Tool registration in the MCP server: tool names are generated as 'spix_{path_with_underscores}' (e.g., 'spix_call_create' from 'call.create'). The `call_tool` handler delegates to `create_tool_handler`.

  # ─── Tool Surface ─────────────────────────────────────────────────────────
  tool_schemas = get_mcp_tools(profile=tool_profile, disabled=disabled_tools)
  tool_defs: list[Tool] = []
  
  for schema in tool_schemas:
      # Convert path to tool name: playbook.create -> spix_playbook_create
      tool_name = f"spix_{schema.path.replace('.', '_')}"
      tool_defs.append(
          Tool(
              name=tool_name,
              description=schema.description or f"Spix {schema.path}",
              inputSchema=build_json_schema(schema),
          )
      )

• src/spix_mcp/registry.py:284-302 (schema)

  The CommandSchema for 'call.create' defines the tool. It has path='call.create', http_method='POST', api_endpoint='/calls', mcp_expose='tool', financial=True. Positional args: 'to' (required). Params: 'playbook_id' (required), 'sender' (required), 'contact_id', 'webhook_url'.

  CommandSchema(
      path="call.create",
      cli_usage="spix call create <to> --playbook <id> --sender <number>",
      http_method="POST",
      api_endpoint="/calls",
      mcp_expose="tool",
      mcp_profile="safe",
      financial=True,
      description="Initiate a voice call",
      positional_args=[
          CommandParam("to", "string", required=True, description="Destination phone number (E.164)"),
      ],
      params=[
          CommandParam("playbook_id", "string", required=True, description="Playbook ID"),
          CommandParam("sender", "string", required=True, description="Sender number (E.164)"),
          CommandParam("contact_id", "string", description="Link to existing contact"),
          CommandParam("webhook_url", "string", description="Override webhook URL"),
      ],
  ),

• src/spix_mcp/tools.py:23-50 (helper)

  The `get_schema_by_tool_name` function maps MCP tool names (like 'spix_call_create') back to command schemas by stripping the 'spix_' prefix and converting underscores to dots for path lookup.

  def get_schema_by_tool_name(tool_name: str) -> CommandSchema | None:
      """Look up a CommandSchema by MCP tool name.
  
      MCP tool names follow the pattern: spix_{path with dots replaced by underscores}
      e.g., "spix_playbook_create" -> "playbook.create"
  
      Args:
          tool_name: The MCP tool name (e.g., "spix_playbook_create").
  
      Returns:
          The matching CommandSchema, or None if not found.
      """
      # Remove the spix_ prefix
      if not tool_name.startswith("spix_"):
          return None
  
      path_part = tool_name[len("spix_") :]
  
      # Convert underscores back to dots for path lookup
      # We need to handle multi-part paths like "billing_credits_history" -> "billing.credits.history"
      # Try different dot positions to find the right one
      for cmd in COMMAND_REGISTRY:
          # Convert the command path to expected tool name format
          expected_tool = cmd.path.replace(".", "_")
          if expected_tool == path_part:
              return cmd
  
      return None

• src/spix_mcp/registry.py:877-922 (helper)

  The `build_json_schema` function converts a CommandSchema into JSON Schema format for MCP tool inputSchema, defining the expected parameters (to, playbook_id, sender, etc.) and their types for the 'spix_call_create' tool.

  def build_json_schema(schema: CommandSchema) -> dict:
      """Convert a CommandSchema into a JSON Schema for MCP tool inputSchema.
  
      Args:
          schema: The command schema to convert.
  
      Returns:
          JSON Schema dict with properties and required fields.
      """
      properties: dict[str, dict] = {}
      required: list[str] = []
  
      type_map = {
          "string": "string",
          "integer": "integer",
          "boolean": "boolean",
          "enum": "string",
          "file": "string",
          "array": "array",
          "number": "number",
          "uuid": "string",  # UUID is represented as string in JSON Schema
      }
  
      for param in schema.positional_args + schema.params:
          prop: dict = {
              "type": type_map.get(param.type, "string"),
          }
          if param.description:
              prop["description"] = param.description
          if param.choices:
              prop["enum"] = param.choices
          if param.default is not None:
              prop["default"] = param.default
          if param.type == "array":
              prop["items"] = {"type": "string"}
  
          properties[param.name] = prop
  
          if param.required:
              required.append(param.name)
  
      return {
          "type": "object",
          "properties": properties,
          "required": required,
      }