diff --git a/lib/stekker_zaptec.rb b/lib/stekker_zaptec.rb
index 5b315fe..9f9c670 100644
--- a/lib/stekker_zaptec.rb
+++ b/lib/stekker_zaptec.rb
@@ -10,9 +10,12 @@
 require "zaptec/errors"
 require "zaptec/installation"
 require "zaptec/installation_hierarchy"
+require "zaptec/messaging_connection_details"
+require "zaptec/messaging_subscription"
 require "zaptec/meter_reading"
 require "zaptec/null_encryptor"
 require "zaptec/state"
+require "zaptec/user_group"
 require "zaptec/version"
 
 module Zaptec
diff --git a/lib/zaptec/client.rb b/lib/zaptec/client.rb
index 5856125..a7102b5 100644
--- a/lib/zaptec/client.rb
+++ b/lib/zaptec/client.rb
@@ -110,6 +110,19 @@ def update_charger(charger_id, **attributes)
       post("/api/chargers/#{charger_id}/update", body: attributes)
     end
 
+    def user_groups
+      response = get("/api/userGroups")
+      data = response.body
+      data = data.fetch("Data") if data.is_a?(Hash) && data.key?("Data")
+      data.map { |attrs| UserGroup.new(attrs) }
+    end
+
+    # https://docs.zaptec.com/docs/zaptec-service-bus-subscriptions
+    def messaging_connection_details(user_group_id)
+      get("/api/userGroups/#{user_group_id}/messagingConnectionDetails")
+        .then { |response| MessagingConnectionDetails.new(response.body) }
+    end
+
     def pause_charging(charger_id) = send_command(charger_id, :StopChargingFinal)
 
     def resume_charging(charger_id) = send_command(charger_id, :ResumeCharging)
diff --git a/lib/zaptec/errors.rb b/lib/zaptec/errors.rb
index 22daf52..5beb400 100644
--- a/lib/zaptec/errors.rb
+++ b/lib/zaptec/errors.rb
@@ -3,14 +3,31 @@ module Errors
     class Base < StandardError; end
     class ParameterMissing < Base; end
     class Forbidden < Base; end
+
     class RequestFailed < Base
       attr_reader :response
 
       def initialize(message, response = nil)
         @response = response
-        super(message)
+        super(format_message(message, response))
+      end
+
+      private
+
+      def format_message(message, response)
+        body = extract_body(response)
+        body.present? ? "#{message}: #{body}" : message
+      end
+
+      def extract_body(response)
+        case response
+        when nil then nil
+        when Hash then response[:body]
+        else response.respond_to?(:body) ? response.body : nil
+        end
       end
     end
+
     class AuthorizationFailed < Base; end
   end
 end
diff --git a/lib/zaptec/messaging_connection_details.rb b/lib/zaptec/messaging_connection_details.rb
new file mode 100644
index 0000000..47dcf85
--- /dev/null
+++ b/lib/zaptec/messaging_connection_details.rb
@@ -0,0 +1,15 @@
+module Zaptec
+  class MessagingConnectionDetails
+    def initialize(data)
+      @data = data.deep_symbolize_keys
+    end
+
+    def host = @data.fetch(:Host)
+    def port = @data.fetch(:Port)
+    def use_ssl = @data.fetch(:UseSSL)
+    def username = @data.fetch(:Username)
+    def password = @data.fetch(:Password)
+    def topic = @data.fetch(:Topic)
+    def subscription = @data.fetch(:Subscription)
+  end
+end
diff --git a/lib/zaptec/messaging_subscription.rb b/lib/zaptec/messaging_subscription.rb
new file mode 100644
index 0000000..a907d1d
--- /dev/null
+++ b/lib/zaptec/messaging_subscription.rb
@@ -0,0 +1,164 @@
+require "base64"
+require "json"
+require "net/http"
+require "openssl"
+require "uri"
+
+module Zaptec
+  class MessagingSubscription
+    Message = Data.define(
+      :device_id,
+      :device_type,
+      :state_id,
+      :state_name,
+      :timestamp,
+      :value,
+      :raw,
+      :lock_location,
+    ) do
+      def initialize(lock_location: nil, **)
+        super
+      end
+    end
+
+    TOKEN_TTL = 1.hour
+    TOKEN_REFRESH_MARGIN = 5.minutes
+
+    def initialize(connection_details:)
+      @connection_details = connection_details
+    end
+
+    def each_message(mode: :consume, timeout: 30)
+      return enum_for(:each_message, mode:, timeout:) unless block_given?
+
+      loop do
+        message = fetch_one(mode:, timeout:)
+        yield message if message
+      end
+    end
+
+    def complete(message)
+      raise ArgumentError, "Message has no lock; was it received in peek mode?" if message.lock_location.nil?
+
+      response = http.request(lock_request(Net::HTTP::Delete, message.lock_location))
+      handle_lock_response(response)
+    end
+
+    private
+
+    attr_reader :connection_details
+
+    def fetch_one(mode:, timeout:)
+      response = http.request(message_request(mode:, timeout:))
+
+      case response
+      when Net::HTTPNoContent
+        nil
+      when Net::HTTPSuccess
+        parse_message(response.body.to_s, lock_location: response["Location"])
+      when Net::HTTPUnauthorized
+        invalidate_token!
+        raise Errors::AuthorizationFailed
+      else
+        raise Errors::RequestFailed.new("#{response.code} #{response.message}: #{response.body}", response)
+      end
+    end
+
+    def message_request(mode:, timeout:)
+      klass = mode == :peek ? Net::HTTP::Post : Net::HTTP::Delete
+      klass.new(message_path(timeout)).tap do |req|
+        req["Authorization"] = sas_token
+        req["Content-Length"] = "0"
+      end
+    end
+
+    def lock_request(klass, lock_location)
+      uri = URI(lock_location)
+      klass.new(uri.request_uri).tap do |req|
+        req["Authorization"] = sas_token
+        req["Content-Length"] = "0"
+      end
+    end
+
+    def handle_lock_response(response)
+      case response
+      when Net::HTTPSuccess
+        nil
+      when Net::HTTPUnauthorized
+        invalidate_token!
+        raise Errors::AuthorizationFailed
+      else
+        raise Errors::RequestFailed.new("#{response.code} #{response.message}: #{response.body}", response)
+      end
+    end
+
+    def message_path(timeout)
+      "/#{connection_details.topic}/subscriptions/#{connection_details.subscription}" \
+        "/messages/head?timeout=#{timeout}"
+    end
+
+    def parse_message(body, lock_location:)
+      start_index = body.index("{")
+      end_index = body.rindex("}")
+      return nil unless start_index && end_index && end_index > start_index
+
+      raw = JSON.parse(body[start_index..end_index])
+      state_id = raw["StateId"]
+      device_type = raw["DeviceType"]
+
+      Message.new(
+        device_id: raw["DeviceId"],
+        device_type:,
+        state_id:,
+        state_name: state_name_for(state_id, device_type),
+        timestamp: raw["Timestamp"],
+        value: raw["ValueAsString"],
+        raw:,
+        lock_location:,
+      )
+    rescue JSON::ParserError
+      nil
+    end
+
+    def state_name_for(state_id, device_type)
+      return unless state_id && device_type
+
+      Constants.observation_state_id_to_name(state_id:, device_type:)
+    end
+
+    def sas_token
+      now = Time.now.to_i
+      if @token.nil? || now > @token_expiry - TOKEN_REFRESH_MARGIN.to_i
+        @token_expiry = now + TOKEN_TTL.to_i
+        @token = build_sas_token(expiry: @token_expiry)
+      end
+      @token
+    end
+
+    def build_sas_token(expiry:)
+      encoded_uri = URI.encode_www_form_component("https://#{connection_details.host}/")
+      signature = Base64.strict_encode64(
+        OpenSSL::HMAC.digest("sha256", connection_details.password, "#{encoded_uri}\n#{expiry}"),
+      )
+
+      "SharedAccessSignature " \
+        "sr=#{encoded_uri}" \
+        "&sig=#{URI.encode_www_form_component(signature)}" \
+        "&se=#{expiry}" \
+        "&skn=#{connection_details.username}"
+    end
+
+    def invalidate_token!
+      @token = nil
+      @token_expiry = nil
+    end
+
+    def http
+      @http ||=
+        Net::HTTP.new(connection_details.host, 443).tap do |conn|
+          conn.use_ssl = true
+          conn.read_timeout = 90
+        end
+    end
+  end
+end
diff --git a/lib/zaptec/user_group.rb b/lib/zaptec/user_group.rb
new file mode 100644
index 0000000..8243a80
--- /dev/null
+++ b/lib/zaptec/user_group.rb
@@ -0,0 +1,10 @@
+module Zaptec
+  class UserGroup
+    def initialize(data)
+      @data = data.deep_symbolize_keys
+    end
+
+    def id = @data.fetch(:Id) { @data.fetch(:id) }
+    def name = @data[:Name] || @data[:name]
+  end
+end
diff --git a/spec/zaptec/client_spec.rb b/spec/zaptec/client_spec.rb
index 394b5d7..d61477f 100644
--- a/spec/zaptec/client_spec.rb
+++ b/spec/zaptec/client_spec.rb
@@ -651,6 +651,77 @@
     end
   end
 
+  describe "#user_groups" do
+    it "returns the user groups accessible to the authenticated user" do
+      WebMock::API
+        .stub_request(:get, "https://api.zaptec.com/api/userGroups")
+        .with(headers: { "Authorization" => "Bearer T123" })
+        .to_return(
+          body: {
+            Data: [
+              { Id: "g1", Name: "Acme" },
+              { Id: "g2", Name: "Beta" },
+            ],
+          }.to_json,
+          headers: { "Content-Type": "application/json" },
+        )
+
+      token_cache = build_token_cache("T123")
+      client = Zaptec::Client.new(username: "zap", password: "tec", token_cache:)
+
+      expect(client.user_groups.map(&:id)).to eq %w[g1 g2]
+      expect(client.user_groups.map(&:name)).to eq %w[Acme Beta]
+    end
+
+    it "supports an unwrapped array response" do
+      WebMock::API
+        .stub_request(:get, "https://api.zaptec.com/api/userGroups")
+        .with(headers: { "Authorization" => "Bearer T123" })
+        .to_return(
+          body: [{ Id: "g1", Name: "Acme" }].to_json,
+          headers: { "Content-Type": "application/json" },
+        )
+
+      token_cache = build_token_cache("T123")
+      client = Zaptec::Client.new(username: "zap", password: "tec", token_cache:)
+
+      expect(client.user_groups.map(&:id)).to eq %w[g1]
+    end
+  end
+
+  describe "#messaging_connection_details" do
+    it "fetches Service Bus connection details for a user group" do
+      WebMock::API
+        .stub_request(:get, "https://api.zaptec.com/api/userGroups/g123/messagingConnectionDetails")
+        .with(headers: { "Authorization" => "Bearer T123" })
+        .to_return(
+          body: {
+            Type: 0,
+            Host: "sb.example.com",
+            Port: 5671,
+            UseSSL: true,
+            Username: "usergroup_g123",
+            Password: "secret",
+            Topic: "usergroup_g123",
+            Subscription: "default",
+          }.to_json,
+          headers: { "Content-Type": "application/json" },
+        )
+
+      token_cache = build_token_cache("T123")
+      client = Zaptec::Client.new(username: "zap", password: "tec", token_cache:)
+
+      expect(client.messaging_connection_details("g123"))
+        .to have_attributes(
+          host: "sb.example.com",
+          username: "usergroup_g123",
+          password: "secret",
+          topic: "usergroup_g123",
+          subscription: "default",
+        )
+    end
+  end
+
   private
 
   def chargers_example
diff --git a/spec/zaptec/errors/request_failed_spec.rb b/spec/zaptec/errors/request_failed_spec.rb
new file mode 100644
index 0000000..5d55c0c
--- /dev/null
+++ b/spec/zaptec/errors/request_failed_spec.rb
@@ -0,0 +1,35 @@
+RSpec.describe Zaptec::Errors::RequestFailed do
+  it "appends the body from a Faraday response hash to the message" do
+    error = described_class.new("Request returned status 503", { body: "Service Unavailable" })
+
+    expect(error.message).to eq("Request returned status 503: Service Unavailable")
+  end
+
+  it "appends the body from a Net::HTTP response object to the message" do
+    response = instance_double(Net::HTTPResponse, body: "<html>nope</html>")
+
+    error = described_class.new("503 Service Unavailable", response)
+
+    expect(error.message).to eq("503 Service Unavailable: <html>nope</html>")
+  end
+
+  it "omits the body separator when no response is given" do
+    error = described_class.new("something broke")
+
+    expect(error.message).to eq("something broke")
+  end
+
+  it "omits the body separator when the body is blank" do
+    error = described_class.new("Request returned status 500", { body: "" })
+
+    expect(error.message).to eq("Request returned status 500")
+  end
+
+  it "exposes the original response object" do
+    response = { status: 503, body: "down" }
+
+    error = described_class.new("boom", response)
+
+    expect(error.response).to eq(response)
+  end
+end
diff --git a/spec/zaptec/messaging_connection_details_spec.rb b/spec/zaptec/messaging_connection_details_spec.rb
new file mode 100644
index 0000000..f636dc9
--- /dev/null
+++ b/spec/zaptec/messaging_connection_details_spec.rb
@@ -0,0 +1,24 @@
+RSpec.describe Zaptec::MessagingConnectionDetails do
+  it "exposes the connection settings" do
+    details = described_class.new(
+      "Type" => 0,
+      "Host" => "zap-p-installations-sbus.servicebus.windows.net",
+      "Port" => 5671,
+      "UseSSL" => true,
+      "Username" => "usergroup_abc",
+      "Password" => "secret",
+      "Topic" => "usergroup_abc",
+      "Subscription" => "default",
+    )
+
+    expect(details).to have_attributes(
+      host: "zap-p-installations-sbus.servicebus.windows.net",
+      port: 5671,
+      use_ssl: true,
+      username: "usergroup_abc",
+      password: "secret",
+      topic: "usergroup_abc",
+      subscription: "default",
+    )
+  end
+end
diff --git a/spec/zaptec/messaging_subscription_spec.rb b/spec/zaptec/messaging_subscription_spec.rb
new file mode 100644
index 0000000..c304824
--- /dev/null
+++ b/spec/zaptec/messaging_subscription_spec.rb
@@ -0,0 +1,166 @@
+RSpec.describe Zaptec::MessagingSubscription do
+  let(:connection_details) do
+    Zaptec::MessagingConnectionDetails.new(
+      "Type" => 0,
+      "Host" => "sb.example.com",
+      "Port" => 5671,
+      "UseSSL" => true,
+      "Username" => "usergroup_abc",
+      "Password" => Base64.strict_encode64("secret"),
+      "Topic" => "usergroup_abc",
+      "Subscription" => "default",
+    )
+  end
+
+  it "yields parsed messages with translated state names" do
+    body = <<~BODY
+      @string\x03http://schemas.microsoft.com/2003/10/Serialization/\xC3\xBE\xC3\xBF{"DeviceId":"ZAP018643","DeviceType":4,"ChargerId":"f94cb4fc-d717-4357-a541-3d54ed815792","StateId":553,"Timestamp":"2026-05-20T12:41:41.55472Z","ValueAsString":"13.463"}
+    BODY
+
+    WebMock::API
+      .stub_request(:delete, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(query: { timeout: "30" })
+      .to_return(status: 201, body:, headers: { "Content-Type" => "application/xml" })
+
+    subscription = described_class.new(connection_details:)
+
+    message = subscription.each_message.first
+
+    expect(message).to have_attributes(
+      device_id: "ZAP018643",
+      device_type: 4,
+      state_id: 553,
+      state_name: :TotalChargePowerSession,
+      timestamp: "2026-05-20T12:41:41.55472Z",
+      value: "13.463",
+      lock_location: nil,
+    )
+  end
+
+  it "captures the Location header as lock_location when peeking" do
+    body = <<~BODY
+      @string\x03http://schemas.microsoft.com/2003/10/Serialization/\xC3\xBE\xC3\xBF{"DeviceId":"ZAP018643","DeviceType":4,"StateId":553,"Timestamp":"2026-05-20T12:41:41.55472Z","ValueAsString":"13.463"}
+    BODY
+
+    WebMock::API
+      .stub_request(:post, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(query: { timeout: "30" })
+      .to_return(
+        status: 201,
+        body:,
+        headers: {
+          "Content-Type" => "application/xml",
+          "Location" => "https://sb.example.com/usergroup_abc/subscriptions/default/messages/42/lock-token",
+        },
+      )
+
+    subscription = described_class.new(connection_details:)
+
+    expect(subscription.send(:fetch_one, mode: :peek, timeout: 30).lock_location)
+      .to eq("https://sb.example.com/usergroup_abc/subscriptions/default/messages/42/lock-token")
+  end
+
+  it "uses POST in peek mode and DELETE in consume mode" do
+    stub_peek = WebMock::API
+      .stub_request(:post, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(query: { timeout: "30" })
+      .to_return(status: 204)
+    stub_consume = WebMock::API
+      .stub_request(:delete, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(query: { timeout: "30" })
+      .to_return(status: 204)
+
+    subscription = described_class.new(connection_details:)
+    subscription.send(:fetch_one, mode: :peek, timeout: 30)
+    subscription.send(:fetch_one, mode: :consume, timeout: 30)
+
+    expect(stub_peek).to have_been_requested
+    expect(stub_consume).to have_been_requested
+  end
+
+  it "sends a SharedAccessSignature Authorization header with the right skn" do
+    stub = WebMock::API
+      .stub_request(:delete, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(
+        query: { timeout: "30" },
+        headers: { "Authorization" => /SharedAccessSignature .+skn=usergroup_abc/ },
+      )
+      .to_return(status: 204)
+
+    described_class.new(connection_details:).send(:fetch_one, mode: :consume, timeout: 30)
+
+    expect(stub).to have_been_requested
+  end
+
+  it "raises AuthorizationFailed on 401" do
+    WebMock::API
+      .stub_request(:delete, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(query: { timeout: "30" })
+      .to_return(status: 401, body: "<Error><Code>401</Code></Error>")
+
+    subscription = described_class.new(connection_details:)
+
+    expect { subscription.send(:fetch_one, mode: :consume, timeout: 30) }
+      .to raise_error(Zaptec::Errors::AuthorizationFailed)
+  end
+
+  it "returns nil on 204 No Content" do
+    WebMock::API
+      .stub_request(:delete, "https://sb.example.com/usergroup_abc/subscriptions/default/messages/head")
+      .with(query: { timeout: "30" })
+      .to_return(status: 204)
+
+    subscription = described_class.new(connection_details:)
+
+    expect(subscription.send(:fetch_one, mode: :consume, timeout: 30)).to be_nil
+  end
+
+  describe "#complete" do
+    it "deletes the locked message at its lock_location" do
+      lock_url = "https://sb.example.com/usergroup_abc/subscriptions/default/messages/42/lock-token"
+      stub = WebMock::API
+        .stub_request(:delete, lock_url)
+        .with(headers: { "Authorization" => /SharedAccessSignature/ })
+        .to_return(status: 200)
+
+      subscription = described_class.new(connection_details:)
+      message = build_message(lock_location: lock_url)
+
+      subscription.complete(message)
+
+      expect(stub).to have_been_requested
+    end
+
+    it "raises ArgumentError when the message was not peek-locked" do
+      subscription = described_class.new(connection_details:)
+      message = build_message(lock_location: nil)
+
+      expect { subscription.complete(message) }.to raise_error(ArgumentError, /peek mode/)
+    end
+
+    it "raises RequestFailed when the broker rejects the completion" do
+      lock_url = "https://sb.example.com/usergroup_abc/subscriptions/default/messages/42/lock-token"
+      WebMock::API
+        .stub_request(:delete, lock_url)
+        .to_return(status: 410, body: "MessageLockLost")
+
+      subscription = described_class.new(connection_details:)
+      message = build_message(lock_location: lock_url)
+
+      expect { subscription.complete(message) }.to raise_error(Zaptec::Errors::RequestFailed, /410/)
+    end
+  end
+
+  def build_message(lock_location:)
+    Zaptec::MessagingSubscription::Message.new(
+      device_id: "ZAP018643",
+      device_type: 4,
+      state_id: 553,
+      state_name: :TotalChargePowerSession,
+      timestamp: "2026-05-20T12:41:41.55472Z",
+      value: "13.463",
+      raw: {},
+      lock_location:,
+    )
+  end
+end
diff --git a/spec/zaptec/user_group_spec.rb b/spec/zaptec/user_group_spec.rb
new file mode 100644
index 0000000..42b5b00
--- /dev/null
+++ b/spec/zaptec/user_group_spec.rb
@@ -0,0 +1,13 @@
+RSpec.describe Zaptec::UserGroup do
+  it "exposes id and name from a Data envelope row" do
+    user_group = described_class.new("Id" => "abc-123", "Name" => "Acme")
+
+    expect(user_group).to have_attributes(id: "abc-123", name: "Acme")
+  end
+
+  it "accepts lowercase keys" do
+    user_group = described_class.new("id" => "abc-123", "name" => "Acme")
+
+    expect(user_group).to have_attributes(id: "abc-123", name: "Acme")
+  end
+end