Header Parameters

Use Header() to declare that a parameter should be read from message headers.

AsyncFast header handling follows the AsyncAPI specification for message headers and schema and is therefore case sensitive

By default, the header name is derived from the argument name and underscores become hyphens, for example, the argument request_id would become request-id

You can also set an explicit header key using alias= (useful for exact casing like Idempotency-Key or ETag). These aliases are reflected both at runtime and in the generated AsyncAPI schema.

Example

from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Header

app = AsyncFast()


@app.channel("order")
async def handle_order(
    idempotency_key: Annotated[str, Header()],
) -> None:
    print(idempotency_key)

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "HandleOrder": {
      "address": "order",
      "messages": {
        "HandleOrderMessage": {
          "$ref": "#/components/messages/HandleOrderMessage"
        }
      }
    }
  },
  "operations": {
    "receiveHandleOrder": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/HandleOrder"
      }
    }
  },
  "components": {
    "messages": {
      "HandleOrderMessage": {
        "headers": {
          "$ref": "#/components/schemas/HandleOrderMessageHeaders"
        }
      }
    },
    "schemas": {
      "HandleOrderMessageHeaders": {
        "properties": {
          "idempotency-key": {
            "title": "Idempotency-Key",
            "type": "string"
          }
        },
        "required": [
          "idempotency-key"
        ],
        "title": "HandleOrderMessageHeaders",
        "type": "object"
      }
    }
  }
}

Header Aliasing

Use alias= when you need a specific header name (including casing):

Example

from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Header

app = AsyncFast()


@app.channel("topic")
async def topic_handler(
    etag: Annotated[str, Header(alias="ETag")],
) -> None:
    print(etag)

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "TopicHandler": {
      "address": "topic",
      "messages": {
        "TopicHandlerMessage": {
          "$ref": "#/components/messages/TopicHandlerMessage"
        }
      }
    }
  },
  "operations": {
    "receiveTopicHandler": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/TopicHandler"
      }
    }
  },
  "components": {
    "messages": {
      "TopicHandlerMessage": {
        "headers": {
          "$ref": "#/components/schemas/TopicHandlerMessageHeaders"
        }
      }
    },
    "schemas": {
      "TopicHandlerMessageHeaders": {
        "properties": {
          "ETag": {
            "title": "Etag",
            "type": "string"
          }
        },
        "required": [
          "ETag"
        ],
        "title": "TopicHandlerMessageHeaders",
        "type": "object"
      }
    }
  }
}

Default

Header parameters can have defaults, just like normal arguments. If the header is not present, the default value is used.

Example

from typing import Annotated

from asyncfast import AsyncFast
from asyncfast import Header

app = AsyncFast()


@app.channel("notification_channel")
async def notification_channel_handler(
    request_id: Annotated[int, Header()] = 0,
) -> None:
    print(f"request_id: {request_id}")

AsyncAPI

{
  "asyncapi": "3.0.0",
  "info": {
    "title": "AsyncFast",
    "version": "0.1.0"
  },
  "channels": {
    "NotificationChannelHandler": {
      "address": "notification_channel",
      "messages": {
        "NotificationChannelHandlerMessage": {
          "$ref": "#/components/messages/NotificationChannelHandlerMessage"
        }
      }
    }
  },
  "operations": {
    "receiveNotificationChannelHandler": {
      "action": "receive",
      "channel": {
        "$ref": "#/channels/NotificationChannelHandler"
      }
    }
  },
  "components": {
    "messages": {
      "NotificationChannelHandlerMessage": {
        "headers": {
          "$ref": "#/components/schemas/NotificationChannelHandlerMessageHeaders"
        }
      }
    },
    "schemas": {
      "NotificationChannelHandlerMessageHeaders": {
        "properties": {
          "request-id": {
            "default": 0,
            "title": "Request-Id",
            "type": "integer"
          }
        },
        "title": "NotificationChannelHandlerMessageHeaders",
        "type": "object"
      }
    }
  }
}