Skip to content

fix: fix sse deserialization #35

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SergeyMenshykh merged 4 commits into from
Jul 8, 2025
Merged

fix: fix sse deserialization #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3ef1fb8
fix sse deserialization
SergeyMenshykh Jul 8, 2025
63038a0
remove unused code
SergeyMenshykh Jul 8, 2025
680f8df
address pr review feedback
SergeyMenshykh Jul 8, 2025
01ae1d0
address pr review feedback
SergeyMenshykh Jul 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions samples/Client/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,8 @@ private static async Task<AgentTask> SendMessageToAgentStreaming(string messageT
{
Message = new Message
{
MessageId = Guid.NewGuid().ToString("N"),
ContextId = currentSessionId,
Role = MessageRole.User,
Parts =
[
Expand Down Expand Up @@ -219,6 +221,18 @@ private static async Task<AgentTask> SendMessageToAgentStreaming(string messageT
Console.WriteLine($"Artifacts count: {agentTask.Artifacts.Count}");
}
break;
case Message message:
taskId = message.TaskId;
Console.WriteLine($"Received message {message.MessageId} with role {message.Role}");
if (message.Parts.Count > 0)
{
Console.WriteLine($"Message parts count: {message.Parts.Count}");
foreach (var part in message.Parts.OfType<TextPart>())
{
Console.WriteLine(part.Text);
}
}
break;
default:
Console.WriteLine($"Unknown event type: {item.EventType}");
break;
Expand Down
21 changes: 17 additions & 4 deletions src/A2A/Client/A2AClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,10 +100,14 @@ private async Task<TOutput> SendRpcRequestAsync<TInput, TOutput>(
"application/json",
cancellationToken).ConfigureAwait(false);

var responseObject = await JsonSerializer.DeserializeAsync(responseStream, A2AJsonUtilities.JsonContext.Default.JsonRpcResponse, cancellationToken) ??
throw new InvalidOperationException("Failed to deserialize the response.");
var responseObject = await JsonSerializer.DeserializeAsync(responseStream, A2AJsonUtilities.JsonContext.Default.JsonRpcResponse, cancellationToken);

return responseObject.Result?.Deserialize(outputTypeInfo) ??
if (responseObject?.Error is { } error)
{
throw new InvalidOperationException($"JSON-RPC error ({error.Code}): {error.Message}");
}

return responseObject?.Result?.Deserialize(outputTypeInfo) ??
throw new InvalidOperationException("Response does not contain a result.");
}

Expand All @@ -124,7 +128,16 @@ private async IAsyncEnumerable<SseItem<TOutput>> SendRpcSseRequestAsync<TInput,
var sseParser = SseParser.Create(responseStream, (eventType, data) =>
{
var reader = new Utf8JsonReader(data);
return JsonSerializer.Deserialize(ref reader, outputTypeInfo) ?? throw new InvalidOperationException("Failed to deserialize the event.");

var responseObject = JsonSerializer.Deserialize(ref reader, A2AJsonUtilities.JsonContext.Default.JsonRpcResponse);

if (responseObject?.Error is { } error)
{
throw new InvalidOperationException($"JSON-RPC error ({error.Code}): {error.Message}");
}

return JsonSerializer.Deserialize(responseObject?.Result, outputTypeInfo) ??
throw new InvalidOperationException("Failed to deserialize the event.");
});

await foreach (var item in sseParser.EnumerateAsync(cancellationToken))
Expand Down
85 changes: 75 additions & 10 deletions tests/A2A.UnitTests/Client/A2AClientTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,8 +379,13 @@ public async Task SendMessageStreamAsync_MapsRequestParamsCorrectly()
};

HttpRequestMessage? capturedRequest = null;
// Simulate a minimal valid SSE response (empty event stream)
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes("event: message\ndata: {}\n\n"));
// Simulate a minimal valid SSE response
var jsonRpcResponse = JsonSerializer.Serialize(new JsonRpcResponse
{
Id = "test-id",
Result = JsonSerializer.SerializeToNode(new { })
});
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes($"event: message\ndata: {jsonRpcResponse}\n\n"));
var response = new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StreamContent(sseStream)
Expand Down Expand Up @@ -436,9 +441,12 @@ public async Task SendMessageStreamAsync_MapsResponseCorrectly()
TaskId = "task-456",
ContextId = "ctx-789"
};
var sseEventJson = JsonSerializer.Serialize<A2AEvent>(expectedMessage, A2AJsonUtilities.DefaultOptions);
var sseData = $"event: message\ndata: {sseEventJson}\n\n";
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes(sseData));
var jsonRpcResponse = JsonSerializer.Serialize(new JsonRpcResponse
{
Id = "test-id",
Result = JsonSerializer.SerializeToNode<A2AEvent>(expectedMessage, A2AJsonUtilities.DefaultOptions)
});
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes($"event: message\ndata: {jsonRpcResponse}\n\n"));
var response = new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StreamContent(sseStream)
Expand Down Expand Up @@ -477,8 +485,13 @@ public async Task ResubscribeToTaskAsync_MapsRequestParamsCorrectly()
// Arrange
var taskId = "task-123";
HttpRequestMessage? capturedRequest = null;
// Simulate a minimal valid SSE response (empty event stream)
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes("event: message\ndata: {}\n\n"));
// Simulate a minimal valid SSE response
var jsonRpcResponse = JsonSerializer.Serialize(new JsonRpcResponse
{
Id = "test-id",
Result = JsonSerializer.SerializeToNode(new { })
});
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes($"event: message\ndata: {jsonRpcResponse}\n\n"));
var response = new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StreamContent(sseStream)
Expand Down Expand Up @@ -520,9 +533,12 @@ public async Task ResubscribeToTaskAsync_MapsResponseCorrectly()
TaskId = "task-456",
ContextId = "ctx-789"
};
var sseEventJson = JsonSerializer.Serialize<A2AEvent>(expectedMessage, A2AJsonUtilities.DefaultOptions);
var sseData = $"event: message\ndata: {sseEventJson}\n\n";
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes(sseData));
var jsonRpcResponse = JsonSerializer.Serialize(new JsonRpcResponse
{
Id = "test-id",
Result = JsonSerializer.SerializeToNode<A2AEvent>(expectedMessage, A2AJsonUtilities.DefaultOptions)
});
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes($"event: message\ndata: {jsonRpcResponse}\n\n"));
var response = new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StreamContent(sseStream)
Expand Down Expand Up @@ -554,6 +570,55 @@ public async Task ResubscribeToTaskAsync_MapsResponseCorrectly()
Assert.Equal(expectedMessage.ContextId, message.ContextId);
}

[Fact]
public async Task SendMessageStreamAsync_ThrowsOnJsonRpcError()
{
// Arrange
var jsonRpcErrorResponse = JsonSerializer.Serialize(JsonRpcResponse.InvalidParamsResponse("test-id"));
var sseStream = new MemoryStream(Encoding.UTF8.GetBytes($"event: message\ndata: {jsonRpcErrorResponse}\n\n"));
var response = new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StreamContent(sseStream)
};
response.Content.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("text/event-stream");
var sut = CreateA2AClient(response);
var sendParams = new MessageSendParams();

// Act & Assert
var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
{
await foreach (var _ in sut.SendMessageStreamAsync(sendParams))
{
// Should throw before yielding any items
}
});

Assert.Contains("-32602", exception.Message);
Assert.Contains("Invalid parameters", exception.Message);
}

[Fact]
public async Task SendMessageAsync_ThrowsOnJsonRpcError()
{
// Arrange
var jsonRpcErrorResponse = JsonSerializer.Serialize(JsonRpcResponse.MethodNotFoundResponse("test-id"));
var response = new HttpResponseMessage(HttpStatusCode.OK)
{
Content = new StringContent(jsonRpcErrorResponse, Encoding.UTF8, "application/json")
};
var sut = CreateA2AClient(response);
var sendParams = new MessageSendParams();

// Act & Assert
var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
{
await sut.SendMessageAsync(sendParams);
});

Assert.Contains("-32601", exception.Message);
Assert.Contains("Method not found", exception.Message);
}

private static A2AClient CreateA2AClient<T>(T result, Action<HttpRequestMessage>? onRequest = null)
{
var jsonResponse = JsonSerializer.Serialize(new JsonRpcResponse
Expand Down