fix(server): improve Swagger /docs example annotations for A2A routes

samples/hello_world_agent.py

@@ -15,6 +15,7 @@
 from a2a.server.events.event_queue import EventQueue
 from a2a.server.request_handlers import DefaultRequestHandler, GrpcHandler
 from a2a.server.routes import (
+    add_a2a_routes_to_fastapi,
     create_agent_card_routes,
     create_jsonrpc_routes,
     create_rest_routes,
@@ -220,9 +221,12 @@ async def serve(
         agent_card=agent_card,
     )
     app = FastAPI()
-    app.routes.extend(jsonrpc_routes)
-    app.routes.extend(agent_card_routes)
-    app.routes.extend(rest_routes)
+    add_a2a_routes_to_fastapi(
+        app,
+        agent_card_routes=agent_card_routes,
+        jsonrpc_routes=jsonrpc_routes,
+        rest_routes=rest_routes,
+    )
 
     grpc_server = grpc.aio.server()
     grpc_server.add_insecure_port(f'{host}:{grpc_port}')

src/a2a/server/routes/_proto_schema.py

@@ -2,6 +2,7 @@
 
 from typing import Any
 
+from google.api import field_behavior_pb2 as _fb
 from google.protobuf.descriptor import Descriptor, FieldDescriptor
 from google.protobuf.message import Message
 
@@ -33,6 +34,15 @@
     FieldDescriptor.TYPE_SINT64: {'type': 'string'},
 }
 
+
+def _is_required(field: FieldDescriptor) -> bool:
+    """Returns True if the field carries google.api.field_behavior = REQUIRED."""
+    try:
+        return _fb.REQUIRED in field.GetOptions().Extensions[_fb.field_behavior]
+    except KeyError:
+        return False
+
+
 _WELL_KNOWN_SCHEMAS: dict[str, dict[str, Any]] = {
     'google.protobuf.Timestamp': {'type': 'string', 'format': 'date-time'},
     'google.protobuf.Duration': {'type': 'string'},
@@ -57,16 +67,44 @@
 
     if field.type == FieldDescriptor.TYPE_MESSAGE:
         item = message_schema(field.message_type, components)
+        # Well-known types return an inline schema (no $ref); don't wrap them as
+        # nullable — they're already inlined as their JSON-Schema equivalent.
+        if not _is_required(field) and '$ref' in item:
+            return {'oneOf': [item, {'type': 'null'}], 'example': None}
     elif field.type == FieldDescriptor.TYPE_ENUM:
-        item = {
-            'type': 'string',
-            'enum': [v.name for v in field.enum_type.values],
-        }
+        values = [v.name for v in field.enum_type.values]
+        example = next(
+            (
+                v
+                for v in values
+                if 'UNSPECIFIED' not in v and 'UNKNOWN' not in v
+            ),
+            values[0] if values else None,
+        )
+        item: dict[str, Any] = {'type': 'string', 'enum': values}
+        if example:
+            item['example'] = example
     else:
         item = dict(_PROTO_SCALAR_SCHEMAS.get(field.type, {'type': 'string'}))
+        if field.type == FieldDescriptor.TYPE_STRING:
+            # REQUIRED fields must be non-empty; use the field name as a
+            # recognisable placeholder. All other strings default to "".
+            item['example'] = field.name if _is_required(field) else ''
+        elif field.type == FieldDescriptor.TYPE_BOOL:
+            item['example'] = False
 
     if field.is_repeated:
-        return {'type': 'array', 'items': item}
+        array_schema: dict[str, Any] = {'type': 'array', 'items': item}
+        # Propagate the item example to the array so Swagger pre-fills one entry
+        # instead of generating one entry per oneOf branch.
+        item_example = (
+            components.get(item['$ref'].split('/')[-1], {}).get('example')
+            if '$ref' in item
+            else item.get('example')
+        )
+        if item_example is not None:
+            array_schema['example'] = [item_example]
+        return array_schema
     return item
 
 
@@ -114,5 +152,14 @@
     if base_properties:
         parts.append({'type': 'object', 'properties': base_properties})
     parts.extend(oneof_constraints)
-    components[name] = parts[0] if len(parts) == 1 else {'allOf': parts}
+    schema: dict[str, Any] = parts[0] if len(parts) == 1 else {'allOf': parts}
+    # Provide a single concrete example using the first oneof variant so Swagger
+    # doesn't expand every branch into separate array items.
+    first_oneof_field = real_oneofs[0].fields[0]
+    schema['example'] = {
+        first_oneof_field.name: first_oneof_field.name
+        if _is_required(first_oneof_field)
+        else ''
+    }
+    components[name] = schema
     return ref

tests/server/routes/test_proto_schema.py

@@ -120,6 +120,79 @@ def test_field_schema_enum():
     assert 'ROLE_AGENT' in schema['enum']
 
 
+def test_field_schema_enum_example_skips_unspecified():
+    role_field = Message.DESCRIPTOR.fields_by_name['role']
+    schema = field_schema(role_field, {})
+    assert schema['example'] == 'ROLE_USER'
+
+
+def test_field_schema_string_example_is_empty():
+    context_id_field = Message.DESCRIPTOR.fields_by_name['context_id']
+    schema = field_schema(context_id_field, {})
+    assert schema['example'] == ''
+
+
+def test_field_schema_string_required_uses_field_name():
+    # REQUIRED string fields must be non-empty; the field name is the placeholder.
+    message_id_field = Message.DESCRIPTOR.fields_by_name['message_id']
+    schema = field_schema(message_id_field, {})
+    assert schema['example'] == 'message_id'
+
+
+def test_field_schema_bool_example_is_false():
+    from a2a.types.a2a_pb2 import SendMessageConfiguration
+
+    field = SendMessageConfiguration.DESCRIPTOR.fields_by_name[
+        'return_immediately'
+    ]
+    schema = field_schema(field, {})
+    assert schema['example'] is False
+
+
+def test_field_schema_optional_message_is_nullable():
+    # Non-REQUIRED message fields default to null so Swagger doesn't pre-fill them
+    # with empty sub-fields that trigger server-side required-field validation.
+    from a2a.types.a2a_pb2 import SendMessageConfiguration
+
+    field = SendMessageConfiguration.DESCRIPTOR.fields_by_name[
+        'task_push_notification_config'
+    ]
+    schema = field_schema(field, {})
+    assert schema['example'] is None
+    assert any(v == {'type': 'null'} for v in schema['oneOf'])
+
+
+def test_field_schema_required_message_is_not_nullable():
+    from a2a.types.a2a_pb2 import SendMessageRequest
+
+    field = SendMessageRequest.DESCRIPTOR.fields_by_name['message']
+    schema = field_schema(field, {})
+    assert '$ref' in schema
+    assert 'oneOf' not in schema
+
+
+def test_message_schema_oneof_example_uses_first_variant_only():
+    components = {}
+    message_schema(Part.DESCRIPTOR, components)
+    example = components['Part']['example']
+    assert example == {'text': ''}
+    # base properties (metadata, filename, media_type) must not appear in the
+    # example — they are objects/strings that would be wrong if sent as "".
+    assert 'metadata' not in example
+    assert 'filename' not in example
+
+
+def test_field_schema_repeated_ref_example_propagated():
+    components = {}
+    msg_descriptor = SendMessageRequest.DESCRIPTOR.fields_by_name[
+        'message'
+    ].message_type
+    parts_field = msg_descriptor.fields_by_name['parts']
+    schema = field_schema(parts_field, components)
+    assert schema['type'] == 'array'
+    assert schema['example'] == [{'text': ''}]
+
+
 def test_field_schema_map_entry():
     metadata_field = SendMessageRequest.DESCRIPTOR.fields_by_name['metadata']
     schema = field_schema(metadata_field, {})