Polytomic MCP Server

Instructions

Implementation Reference

• src/polytomic_mcp/server.py:262-300 (handler)

  The implementation of the create_sync tool, which creates a new model sync in Polytomic.

  async def create_sync(
      name: str,
      mode: str,
      target: str,
      fields: str,
      identity: str | None = None,
      schedule: str | None = None,
      filters: str | None = None,
      filter_logic: str | None = None,
  ) -> str:
      """Create a new model sync (reverse ETL) in Polytomic.
  
      Args:
          name: Name for the sync
          mode: Sync mode ('create', 'update', 'updateOrCreate', 'replace', 'append')
          target: JSON string with target config {"connection_id": "...", "object": "...", "configuration": {...}}
          fields: JSON array of field mappings [{"source": {"field": "...", "model_id": "..."}, "target": "..."}]
          identity: Optional JSON for identity mapping (required for update modes)
          schedule: Optional JSON for schedule {"frequency": "manual|hourly|daily|weekly|monthly", ...}
          filters: Optional JSON array of filters
          filter_logic: Optional filter logic string (e.g. "A AND B")
      """
      body = {
          "name": name,
          "mode": mode,
          "target": json.loads(target),
          "fields": json.loads(fields),
      }
      if identity:
          body["identity"] = json.loads(identity)
      if schedule:
          body["schedule"] = json.loads(schedule)
      if filters:
          body["filters"] = json.loads(filters)
      if filter_logic:
          body["filter_logic"] = filter_logic
  
      result = await polytomic_request("/syncs", method="POST", body=body)
      return json.dumps(result, indent=2)