Module: A2A::TestHelpers

Defined in:
lib/a2a/test_helpers.rb

Class Method Summary collapse

Class Method Details

.stub_agent(agent_card: {}) ⇒ Object

Returns a stub agent server that handles all A2A operations with minimal valid responses. Useful for integration tests that need a working server without real business logic.



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/a2a/test_helpers.rb', line 13

def self.stub_agent(agent_card: {})
  schema = A2A::Protocol::JsonSchema

  A2A.agent(agent_card: agent_card) do |env|
    case env["a2a.operation"]
    in "SendMessage"
      schema["Send Message Response"].new({})
    in "SendStreamingMessage" | "SubscribeToTask"
      schema["Stream Response"].new({})
    in "GetTask"
      schema["Task"].new(
        "id"        => "test-id",
        "contextId" => "ctx-1",
        "status"    => { "state" => "TASK_STATE_COMPLETED", "timestamp" => "2025-01-01T00:00:00Z" }
      )
    in "CancelTask"
      schema["Task"].new(
        "id"        => "test-id",
        "contextId" => "ctx-1",
        "status"    => { "state" => "TASK_STATE_CANCELED", "timestamp" => "2025-01-01T00:00:00Z" }
      )
    in "ListTasks"
      schema["List Tasks Response"].new({})
    in "CreateTaskPushNotificationConfig" | "GetTaskPushNotificationConfig"
      schema["Task Push Notification Config"].new("url" => "http://example.com")
    in "ListTaskPushNotificationConfigs"
      schema["List Task Push Notification Configs Response"].new({})
    in "GetExtendedAgentCard"
      schema["Agent Card"].new("name" => "Test", "version" => "1.0")
    in "DeleteTaskPushNotificationConfig"
      nil
    end
  end
end