diff --git a/src/service.js b/src/service.js
index 32b07d5..d433e20 100644
--- a/src/service.js
+++ b/src/service.js
@@ -28,6 +28,27 @@ module.exports = function(mixinOptions) {
 	});
 
 	const serviceSchema = {
+		actions: {
+			ws: {
+				visibility: "private",
+				tracing: {
+					tags: {
+						params: ["socket.upgradeReq.url"],
+					},
+					spanName: ctx => `UPGRADE ${ctx.params.socket.upgradeReq.url}`,
+				},
+				handler(ctx) {
+					const { socket, connectionParams } = ctx.params;
+					return {
+						$ctx: ctx,
+						$socket: socket,
+						$service: this,
+						$params: { body: connectionParams, query: socket.upgradeReq.query },
+					};
+				},
+			},
+		},
+
 		events: {
 			"$services.changed"() {
 				if (mixinOptions.autoUpdateSchema) {
@@ -303,14 +324,14 @@ module.exports = function(mixinOptions) {
 					subscribe: filter
 						? withFilter(
 								() => this.pubsub.asyncIterator(tags),
-								async (payload, params, ctx) =>
+								async (payload, params, { ctx }) =>
 									payload !== undefined
-										? this.broker.call(filter, { ...params, payload }, ctx)
+										? ctx.call(filter, { ...params, payload })
 										: false
 						  )
 						: () => this.pubsub.asyncIterator(tags),
-					resolve: async (payload, params, ctx) =>
-						this.broker.call(actionName, { ...params, payload }, ctx),
+					resolve: (payload, params, { ctx }) =>
+						ctx.call(actionName, { ...params, payload }),
 				};
 			},
 
@@ -586,23 +607,23 @@ module.exports = function(mixinOptions) {
 					this.apolloServer = new ApolloServer({
 						schema,
 						..._.defaultsDeep({}, mixinOptions.serverOptions, {
-							context: ({ req, connection }) => {
-								return req
+							context: ({ req, connection }) => ({
+								...(req
 									? {
 											ctx: req.$ctx,
 											service: req.$service,
 											params: req.$params,
-											dataLoaders: new Map(), // create an empty map to load DataLoader instances into
 									  }
 									: {
-											service: connection.$service,
-									  };
-							},
+											ctx: connection.context.$ctx,
+											service: connection.context.$service,
+											params: connection.context.$params,
+									  }),
+								dataLoaders: new Map(), // create an empty map to load DataLoader instances into
+							}),
 							subscriptions: {
-								onConnect: connectionParams => ({
-									...connectionParams,
-									$service: this,
-								}),
+								onConnect: (connectionParams, socket) =>
+									this.actions.ws({ connectionParams, socket }),
 							},
 						}),
 					});
@@ -724,6 +745,7 @@ module.exports = function(mixinOptions) {
 
 	if (mixinOptions.createAction) {
 		serviceSchema.actions = {
+			...serviceSchema.actions,
 			graphql: {
 				params: {
 					query: { type: "string" },
diff --git a/test/unit/service.spec.js b/test/unit/service.spec.js
index e915129..b5b8e64 100644
--- a/test/unit/service.spec.js
+++ b/test/unit/service.spec.js
@@ -965,11 +965,12 @@ describe("Test Service", () => {
 
 			// Test resolve
 			const ctx = new Context(broker);
-			const res3 = await res.resolve({ a: 5 }, { b: "John" }, ctx);
+			ctx.call = jest.fn(async () => "action response");
+			const res3 = await res.resolve({ a: 5 }, { b: "John" }, { ctx });
 
 			expect(res3).toBe("action response");
-			expect(broker.call).toBeCalledTimes(1);
-			expect(broker.call).toBeCalledWith("posts.find", { b: "John", payload: { a: 5 } }, ctx);
+			expect(ctx.call).toBeCalledTimes(1);
+			expect(ctx.call).toBeCalledWith("posts.find", { b: "John", payload: { a: 5 } });
 		});
 
 		it("should create resolver with tags", async () => {
@@ -1003,24 +1004,23 @@ describe("Test Service", () => {
 			});
 
 			// Test first function
-			expect(res.subscribe[0]()).toBe("iterator-result");
+			const ctx = new Context(broker);
+			expect(res.subscribe[0](undefined, undefined, { ctx })).toBe("iterator-result");
 
 			expect(svc.pubsub.asyncIterator).toBeCalledTimes(1);
 			expect(svc.pubsub.asyncIterator).toBeCalledWith(["a", "b"]);
 
 			// Test second function without payload
-			expect(await res.subscribe[1]()).toBe(false);
+			expect(await res.subscribe[1](undefined, undefined, { ctx })).toBe(false);
 
 			// Test second function with payload
-			const ctx = new Context(broker);
-			expect(await res.subscribe[1]({ a: 5 }, { b: "John" }, ctx)).toBe("action response");
-
-			expect(broker.call).toBeCalledTimes(1);
-			expect(broker.call).toBeCalledWith(
-				"posts.filter",
-				{ b: "John", payload: { a: 5 } },
-				ctx
+			ctx.call = jest.fn(async () => "action response");
+			expect(await res.subscribe[1]({ a: 5 }, { b: "John" }, { ctx })).toBe(
+				"action response"
 			);
+
+			expect(ctx.call).toBeCalledTimes(1);
+			expect(ctx.call).toBeCalledWith("posts.filter", { b: "John", payload: { a: 5 } });
 		});
 	});
 
@@ -1442,10 +1442,19 @@ describe("Test Service", () => {
 			expect(
 				contextFn({
 					connection: {
-						$service: "service",
+						context: {
+							$service: "service",
+							$ctx: "context",
+							$params: { a: 5 },
+						},
 					},
 				})
 			).toEqual({
+				ctx: "context",
+				dataLoaders: new Map(),
+				params: {
+					a: 5,
+				},
 				service: "service",
 			});
 
@@ -1474,11 +1483,13 @@ describe("Test Service", () => {
 			const onConnect = ApolloServer.mock.calls[0][0].subscriptions.onConnect;
 
 			const connectionParams = { b: 100 };
+			const socket = { connectionParams, upgradeReq: { query: 101 } };
+			const connect = await onConnect(connectionParams, socket);
 
-			expect(onConnect(connectionParams)).toEqual({
-				b: 100,
-				$service: svc,
-			});
+			expect(connect.$service).toEqual(svc);
+			expect(connect.$ctx).toBeDefined();
+			expect(connect.$params.body).toEqual(connectionParams);
+			expect(connect.$params.query).toEqual(socket.upgradeReq.query);
 
 			await stop();
 		});