Spix

Instructions

Implementation Reference

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

  The create_tool_handler function is the general handler for all MCP tools, including 'spix_playbook_create'. It resolves the tool name to a schema, validates session/access, builds the endpoint URL, dispatches the request using the session client, and formats 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/registry.py:57-86 (schema)

  The CommandSchema definition for 'playbook.create', which is the underlying command for the MCP tool 'spix_playbook_create'. It defines the HTTP method (POST), endpoint (/playbooks), and required parameters.

  CommandSchema(
      path="playbook.create",
      cli_usage="spix playbook create --type <call|sms> --name <n>",
      http_method="POST",
      api_endpoint="/playbooks",
      mcp_expose="tool",
      mcp_profile="safe",
      description="Create a new call or SMS playbook",
      params=[
          CommandParam("type", "enum", required=True, choices=["call", "sms"], description="Playbook type"),
          CommandParam("name", "string", required=True, description="Playbook name"),
          CommandParam("goal", "string", description="Call playbook goal"),
          CommandParam("persona", "string", description="AI persona description"),
          CommandParam("briefing", "string", description="Playbook briefing text"),
          CommandParam("context", "string", description="Playbook context"),
          CommandParam("voice_id", "uuid", description="Voice UUID for TTS"),
          CommandParam("language", "string", description="Language code (e.g. en, es, fr)"),
          CommandParam("default_emotion", "string", description="Default TTS emotion"),
          CommandParam("speaking_rate", "number", description="Speaking rate (0.6-1.5)"),
          CommandParam("tts_volume", "number", description="TTS volume (0.5-2.0)"),
          CommandParam("max_duration_sec", "integer", description="Max call duration in seconds"),
          CommandParam("record", "boolean", default=True, description="Record calls"),
          CommandParam(
              "amd_action",
              "enum",
              choices=["hang_up", "voicemail_leave_message", "retry_later"],
              description="Answering machine detection action",
          ),
      ],
  ),