feat: jsonrpc server (a2aproject#91)

* Implemented `jsonrpc` server transport exposed as
`a2asrv.NewJSONRPCHandler(requestHandler)`.
* Created an `e2e` package for tests using real dependencies (except
`AgentExecutor`).
* Extracted `mockAgentExecutor` to `internal/testutil/testexecutor`
package, unfortunately can't be used in `a2asrv` because of a circular
reference. Will need to think about reorganizing tests later.
* Extracted server-sent-events utilities to `internal/sse` package.
* Extracted jsonrpc-related constants and jsonrpc error mapping to
`internal/jsonrpc` to reuse between `a2asrv` and `a2aclient`.

Author: yarolegovich

a2aclient/auth_test.go

@@ -24,27 +24,13 @@ import (
 	"github.com/a2aproject/a2a-go/a2agrpc"
 	"github.com/a2aproject/a2a-go/a2asrv"
 	"github.com/a2aproject/a2a-go/a2asrv/eventqueue"
+	"github.com/a2aproject/a2a-go/internal/testutil/testexecutor"
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/test/bufconn"
 )
 
-type mockAgentExecutor struct {
-	ExecuteFn func(context.Context, *a2asrv.RequestContext, eventqueue.Queue) error
-}
-
-func (e *mockAgentExecutor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
-	if e.ExecuteFn != nil {
-		return e.ExecuteFn(ctx, reqCtx, q)
-	}
-	return nil
-}
-
-func (e *mockAgentExecutor) Cancel(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
-	return nil
-}
-
 func startGRPCTestServer(t *testing.T, handler a2asrv.RequestHandler, listener *bufconn.Listener) {
 	s := grpc.NewServer()
 	grpcHandler := a2agrpc.NewHandler(handler)
@@ -68,12 +54,10 @@ func TestAuth_GRPC(t *testing.T) {
 	listener := bufconn.Listen(1024 * 1024)
 
 	var capturedCallContext *a2asrv.CallContext
-	executor := &mockAgentExecutor{
-		ExecuteFn: func(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
-			capturedCallContext, _ = a2asrv.CallContextFrom(ctx)
-			return q.Write(ctx, a2a.NewMessage(a2a.MessageRoleAgent))
-		},
-	}
+	executor := testexecutor.FromFunction(func(ctx context.Context, reqCtx *a2asrv.RequestContext, q eventqueue.Queue) error {
+		capturedCallContext, _ = a2asrv.CallContextFrom(ctx)
+		return q.Write(ctx, a2a.NewMessage(a2a.MessageRoleAgent))
+	})
 	handler := a2asrv.NewHandler(executor)
 	go startGRPCTestServer(t, handler, listener)
 

a2aclient/jsonrpc.go

@@ -15,7 +15,6 @@
 package a2aclient
 
 import (
-	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -26,33 +25,27 @@ import (
 	"time"
 
 	"github.com/a2aproject/a2a-go/a2a"
+	"github.com/a2aproject/a2a-go/internal/jsonrpc"
+	"github.com/a2aproject/a2a-go/internal/sse"
 	"github.com/a2aproject/a2a-go/log"
 	"github.com/google/uuid"
 )
 
-// JSON-RPC 2.0 protocol constants
-const (
-	jsonrpcVersion = "2.0"
-
-	// HTTP headers
-	contentTypeJSON   = "application/json"
-	acceptEventStream = "text/event-stream"
-
-	// JSON-RPC method names per A2A spec §7
-	methodMessageSend              = "message/send"
-	methodMessageStream            = "message/stream"
-	methodTasksGet                 = "tasks/get"
-	methodTasksCancel              = "tasks/cancel"
-	methodTasksResubscribe         = "tasks/resubscribe"
-	methodPushConfigGet            = "tasks/pushNotificationConfig/get"
-	methodPushConfigSet            = "tasks/pushNotificationConfig/set"
-	methodPushConfigList           = "tasks/pushNotificationConfig/list"
-	methodPushConfigDelete         = "tasks/pushNotificationConfig/delete"
-	methodGetAuthenticatedExtended = "agent/getAuthenticatedExtendedCard"
-
-	// SSE data prefix
-	sseDataPrefix = "data: "
-)
+// jsonrpcRequest represents a JSON-RPC 2.0 request.
+type jsonrpcRequest struct {
+	JSONRPC string `json:"jsonrpc"`
+	Method  string `json:"method"`
+	Params  any    `json:"params,omitempty"`
+	ID      string `json:"id"`
+}
+
+// jsonrpcResponse represents a JSON-RPC 2.0 response.
+type jsonrpcResponse struct {
+	JSONRPC string          `json:"jsonrpc"`
+	ID      string          `json:"id"`
+	Result  json.RawMessage `json:"result,omitempty"`
+	Error   *jsonrpc.Error  `json:"error,omitempty"`
+}
 
 // JSONRPCOption configures optional parameters for the JSONRPC transport.
 // Options are applied during NewJSONRPCTransport initialization.
@@ -87,18 +80,17 @@ func WithJSONRPCTransport(opts ...JSONRPCOption) FactoryOption {
 	return WithTransport(
 		a2a.TransportProtocolJSONRPC,
 		TransportFactoryFn(func(ctx context.Context, url string, card *a2a.AgentCard) (Transport, error) {
-			return NewJSONRPCTransport(url, card, opts...), nil
+			return NewJSONRPCTransport(url, opts...), nil
 		}),
 	)
 }
 
 // NewJSONRPCTransport creates a new JSON-RPC transport for A2A protocol communication.
 // By default, an HTTP client with 5-second timeout is used (matching Python SDK behavior).
 // For custom timeout, retry logic, or connection pooling, provide a configured client via WithHTTPClient.
-func NewJSONRPCTransport(url string, card *a2a.AgentCard, opts ...JSONRPCOption) Transport {
+func NewJSONRPCTransport(url string, opts ...JSONRPCOption) Transport {
 	t := &jsonrpcTransport{
-		url:       url,
-		agentCard: card,
+		url: url,
 		httpClient: &http.Client{
 			Timeout: 5 * time.Second, // Match Python SDK httpx.AsyncClient default
 		},
@@ -115,45 +107,11 @@ func NewJSONRPCTransport(url string, card *a2a.AgentCard, opts ...JSONRPCOption)
 type jsonrpcTransport struct {
 	url        string
 	httpClient *http.Client
-	agentCard  *a2a.AgentCard
-}
-
-// jsonrpcRequest represents a JSON-RPC 2.0 request.
-type jsonrpcRequest struct {
-	JSONRPC string `json:"jsonrpc"`
-	Method  string `json:"method"`
-	Params  any    `json:"params,omitempty"`
-	ID      string `json:"id"`
-}
-
-// jsonrpcResponse represents a JSON-RPC 2.0 response.
-type jsonrpcResponse struct {
-	JSONRPC string          `json:"jsonrpc"`
-	ID      string          `json:"id"`
-	Result  json.RawMessage `json:"result,omitempty"`
-	Error   *jsonrpcError   `json:"error,omitempty"`
-}
-
-// jsonrpcError represents a JSON-RPC 2.0 error object.
-// TODO(yarolegovich): Convert to transport-agnostic error format so Client can use errors.Is(err, a2a.ErrMethodNotFound).
-// This needs to be implemented across all transports (currently not in grpc either).
-type jsonrpcError struct {
-	Code    int             `json:"code"`
-	Message string          `json:"message"`
-	Data    json.RawMessage `json:"data,omitempty"`
-}
-
-// Error implements the error interface for jsonrpcError.
-func (e *jsonrpcError) Error() string {
-	if len(e.Data) > 0 {
-		return fmt.Sprintf("jsonrpc error %d: %s (data: %s)", e.Code, e.Message, string(e.Data))
-	}
-	return fmt.Sprintf("jsonrpc error %d: %s", e.Code, e.Message)
 }
 
 func (t *jsonrpcTransport) newHTTPRequest(ctx context.Context, method string, params any) (*http.Request, error) {
 	req := jsonrpcRequest{
-		JSONRPC: jsonrpcVersion,
+		JSONRPC: jsonrpc.Version,
 		Method:  method,
 		Params:  params,
 		ID:      uuid.New().String(),
@@ -169,7 +127,7 @@ func (t *jsonrpcTransport) newHTTPRequest(ctx context.Context, method string, pa
 		return nil, fmt.Errorf("failed to create HTTP request: %w", err)
 	}
 
-	httpReq.Header.Set("Content-Type", contentTypeJSON)
+	httpReq.Header.Set("Content-Type", jsonrpc.ContentJSON)
 
 	if callMeta, ok := CallMetaFrom(ctx); ok {
 		for k, vals := range callMeta {
@@ -209,7 +167,7 @@ func (t *jsonrpcTransport) sendRequest(ctx context.Context, method string, param
 	}
 
 	if resp.Error != nil {
-		return nil, resp.Error
+		return nil, resp.Error.ToA2AError()
 	}
 
 	return resp.Result, nil
@@ -221,7 +179,7 @@ func (t *jsonrpcTransport) sendStreamingRequest(ctx context.Context, method stri
 	if err != nil {
 		return nil, fmt.Errorf("failed to create HTTP request: %w", err)
 	}
-	httpReq.Header.Set("Accept", acceptEventStream)
+	httpReq.Header.Set("Accept", sse.ContentEventStream)
 
 	httpResp, err := t.httpClient.Do(httpReq)
 	if err != nil {
@@ -241,43 +199,30 @@ func (t *jsonrpcTransport) sendStreamingRequest(ctx context.Context, method stri
 // parseSSEStream parses Server-Sent Events and yields JSON-RPC responses.
 func parseSSEStream(body io.Reader) iter.Seq2[json.RawMessage, error] {
 	return func(yield func(json.RawMessage, error) bool) {
-		scanner := bufio.NewScanner(body)
-		prefixBytes := []byte(sseDataPrefix)
-
-		for scanner.Scan() {
-			lineBytes := scanner.Bytes()
-
-			// SSE data lines start with "data: "
-			if bytes.HasPrefix(lineBytes, prefixBytes) {
-				data := lineBytes[len(prefixBytes):]
-
-				var resp jsonrpcResponse
-				if err := json.Unmarshal(data, &resp); err != nil {
-					yield(nil, fmt.Errorf("failed to parse SSE data: %w", err))
-					return
-				}
-
-				if resp.Error != nil {
-					yield(nil, resp.Error)
-					return
-				}
-
-				if !yield(resp.Result, nil) {
-					return
-				}
+		for data, err := range sse.ParseDataStream(body) {
+			if err != nil {
+				yield(nil, err)
+				return
+			}
+			var resp jsonrpcResponse
+			if err := json.Unmarshal(data, &resp); err != nil {
+				yield(nil, fmt.Errorf("failed to parse SSE data: %w", err))
+				return
+			}
+			if resp.Error != nil {
+				yield(nil, resp.Error.ToA2AError())
+				return
+			}
+			if !yield(resp.Result, nil) {
+				return
 			}
-			// Ignore empty lines, comments, and other SSE event types
-		}
-
-		if err := scanner.Err(); err != nil {
-			yield(nil, fmt.Errorf("SSE stream error: %w", err))
 		}
 	}
 }
 
 // SendMessage sends a non-streaming message to the agent.
 func (t *jsonrpcTransport) SendMessage(ctx context.Context, message *a2a.MessageSendParams) (a2a.SendMessageResult, error) {
-	result, err := t.sendRequest(ctx, methodMessageSend, message)
+	result, err := t.sendRequest(ctx, jsonrpc.MethodMessageSend, message)
 	if err != nil {
 		return nil, err
 	}
@@ -335,12 +280,12 @@ func (t *jsonrpcTransport) streamRequestToEvents(ctx context.Context, method str
 
 // SendStreamingMessage sends a streaming message to the agent.
 func (t *jsonrpcTransport) SendStreamingMessage(ctx context.Context, message *a2a.MessageSendParams) iter.Seq2[a2a.Event, error] {
-	return t.streamRequestToEvents(ctx, methodMessageStream, message)
+	return t.streamRequestToEvents(ctx, jsonrpc.MethodMessageStream, message)
 }
 
 // GetTask retrieves the current state of a task.
 func (t *jsonrpcTransport) GetTask(ctx context.Context, query *a2a.TaskQueryParams) (*a2a.Task, error) {
-	result, err := t.sendRequest(ctx, methodTasksGet, query)
+	result, err := t.sendRequest(ctx, jsonrpc.MethodTasksGet, query)
 	if err != nil {
 		return nil, err
 	}
@@ -355,7 +300,7 @@ func (t *jsonrpcTransport) GetTask(ctx context.Context, query *a2a.TaskQueryPara
 
 // CancelTask requests cancellation of a task.
 func (t *jsonrpcTransport) CancelTask(ctx context.Context, id *a2a.TaskIDParams) (*a2a.Task, error) {
-	result, err := t.sendRequest(ctx, methodTasksCancel, id)
+	result, err := t.sendRequest(ctx, jsonrpc.MethodTasksCancel, id)
 	if err != nil {
 		return nil, err
 	}
@@ -370,12 +315,12 @@ func (t *jsonrpcTransport) CancelTask(ctx context.Context, id *a2a.TaskIDParams)
 
 // ResubscribeToTask reconnects to an SSE stream for an ongoing task.
 func (t *jsonrpcTransport) ResubscribeToTask(ctx context.Context, id *a2a.TaskIDParams) iter.Seq2[a2a.Event, error] {
-	return t.streamRequestToEvents(ctx, methodTasksResubscribe, id)
+	return t.streamRequestToEvents(ctx, jsonrpc.MethodTasksResubscribe, id)
 }
 
 // GetTaskPushConfig retrieves the push notification configuration for a task.
 func (t *jsonrpcTransport) GetTaskPushConfig(ctx context.Context, params *a2a.GetTaskPushConfigParams) (*a2a.TaskPushConfig, error) {
-	result, err := t.sendRequest(ctx, methodPushConfigGet, params)
+	result, err := t.sendRequest(ctx, jsonrpc.MethodPushConfigGet, params)
 	if err != nil {
 		return nil, err
 	}
@@ -390,7 +335,7 @@ func (t *jsonrpcTransport) GetTaskPushConfig(ctx context.Context, params *a2a.Ge
 
 // ListTaskPushConfig lists push notification configurations.
 func (t *jsonrpcTransport) ListTaskPushConfig(ctx context.Context, params *a2a.ListTaskPushConfigParams) ([]*a2a.TaskPushConfig, error) {
-	result, err := t.sendRequest(ctx, methodPushConfigList, params)
+	result, err := t.sendRequest(ctx, jsonrpc.MethodPushConfigList, params)
 	if err != nil {
 		return nil, err
 	}
@@ -405,7 +350,7 @@ func (t *jsonrpcTransport) ListTaskPushConfig(ctx context.Context, params *a2a.L
 
 // SetTaskPushConfig sets or updates the push notification configuration for a task.
 func (t *jsonrpcTransport) SetTaskPushConfig(ctx context.Context, params *a2a.TaskPushConfig) (*a2a.TaskPushConfig, error) {
-	result, err := t.sendRequest(ctx, methodPushConfigSet, params)
+	result, err := t.sendRequest(ctx, jsonrpc.MethodPushConfigSet, params)
 	if err != nil {
 		return nil, err
 	}
@@ -420,16 +365,22 @@ func (t *jsonrpcTransport) SetTaskPushConfig(ctx context.Context, params *a2a.Ta
 
 // DeleteTaskPushConfig deletes a push notification configuration.
 func (t *jsonrpcTransport) DeleteTaskPushConfig(ctx context.Context, params *a2a.DeleteTaskPushConfigParams) error {
-	_, err := t.sendRequest(ctx, methodPushConfigDelete, params)
+	_, err := t.sendRequest(ctx, jsonrpc.MethodPushConfigDelete, params)
 	return err
 }
 
 // GetAgentCard retrieves the agent's card.
 func (t *jsonrpcTransport) GetAgentCard(ctx context.Context) (*a2a.AgentCard, error) {
-	if t.agentCard == nil {
-		return nil, fmt.Errorf("no agent card available")
+	result, err := t.sendRequest(ctx, jsonrpc.MethodGetExtendedAgentCard, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var card a2a.AgentCard
+	if err := json.Unmarshal(result, &card); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal agent card: %w", err)
 	}
-	return t.agentCard, nil
+	return &card, nil
 }
 
 // Destroy closes the transport and releases resources.