aw.routing

Configurable routing and decision chains for agentic workflows.

This module provides tools for building extensible, inspectable routing chains that follow the open-closed principle. Users can: - Use default configurations out of the box - Inspect all routing rules and mappings - Customize individual components - Extend with new routing strategies

The key pattern is: provide sensible defaults, make them visible, keep them mutable.

class aw.routing.ConditionalRouter(strategies: list[~aw.routing.RoutingStrategy] = <factory>, short_circuit: ~typing.Callable[[~typing.Any], bool] | None = None)[source]

Router with short-circuit conditions.

Only accepts results that match the short-circuit condition, otherwise continues to next strategy.

>>> def try_url(x): return x.get('url_ext')
>>> def try_type(x): return x.get('content_type_ext')
>>>
>>> # Only accept .pdf or .md, otherwise keep trying
>>> router = ConditionalRouter(
...     [try_url, try_type],
...     short_circuit=lambda r: r in {'.pdf', '.md'}
... )
>>>
>>> # .pdf matches short-circuit, return immediately
>>> router({'url_ext': '.pdf', 'content_type_ext': '.html'})
'.pdf'
>>>
>>> # .jpg doesn't match, try next strategy, .md matches
>>> router({'url_ext': '.jpg', 'content_type_ext': '.md'})
'.md'

class aw.routing.ExtensionContext(url: str, content: bytes = b'', content_type: str = '', explicit_extension: str | None = None)[source]

Context for extension detection.

>>> ctx = ExtensionContext(
...     url='https://example.com/file.pdf',
...     content=b'%PDF-1.5',
...     content_type='text/html'
... )
>>> ctx.url
'https://example.com/file.pdf'

class aw.routing.ExtensionRouter(priority_extensions: frozenset = <factory>, content_type_map: dict = <factory>, magic_bytes_map: dict = <factory>)[source]

Route extension detection through configurable strategies.

This router implements the default logic for detecting file extensions, but every component is visible and customizable.

Examples

>>> # Use with defaults
>>> router = ExtensionRouter()
>>> ctx = ExtensionContext('https://example.com/file.pdf', b'%PDF-1.5')
>>> router(ctx)
'.pdf'
>>>
>>> # URL without extension, falls back to magic bytes
>>> ctx = ExtensionContext('https://example.com/download', b'%PDF-1.5')
>>> router(ctx)
'.pdf'
>>>
>>> # Priority extensions short-circuit
>>> ctx = ExtensionContext('https://example.com/file.pdf', b'<html>')
>>> router(ctx)  # .pdf from URL wins even though content is HTML
'.pdf'
>>>
>>> # Customize content type mapping
>>> router.content_type_map['.txt'] = 'text/x-log'
>>>
>>> # Add custom detector
>>> def my_detector(ctx): return '.custom' if 'special' in ctx.url else None
>>> router = router.with_prepended_strategy(my_detector)

with_appended_strategy(strategy: Callable[[ExtensionContext], str | None]) → ExtensionRouter[source]: Create new router with added strategy at lowest priority.

with_prepended_strategy(strategy: Callable[[ExtensionContext], str | None]) → ExtensionRouter[source]: Create new router with added strategy at highest priority.

class aw.routing.MappingRouter(mapping: dict = <factory>, default: ~typing.Any = None, key_transform: ~typing.Callable[[~typing.Any], ~typing.Any] | None = None)[source]

Router based on a lookup mapping.

Provides a clean interface for switch-case style routing with: - Visible, mutable mapping - Optional default value or factory - Easy extension via dict interface

>>> # Content-Type to extension mapping
>>> ct_map = {
...     'application/pdf': '.pdf',
...     'text/html': '.html',
...     'application/json': '.json',
... }
>>> router = MappingRouter(ct_map)
>>>
>>> router('application/pdf')
'.pdf'
>>> router('text/plain') is None  # No default
True
>>>
>>> # With default
>>> router_with_default = MappingRouter(ct_map, default='.bin')
>>> router_with_default('text/plain')
'.bin'

copy() → MappingRouter[source]: Create a copy with same configuration.

update(*args, **kwargs)[source]: Update mapping.

class aw.routing.PriorityRouter(strategies: list[RoutingStrategy] = <factory>)[source]

Route through strategies in priority order, short-circuiting on first match.

This is a simpler, more specialized version of i2.routing_forest that’s optimized for the common case of “try these in order, stop at first success”.

>>> def try_a(x): return 'A' if x > 10 else None
>>> def try_b(x): return 'B' if x > 5 else None
>>> def try_c(x): return 'C'
>>>
>>> router = PriorityRouter([try_a, try_b, try_c])
>>> router(15)
'A'
>>> router(8)
'B'
>>> router(3)
'C'

append(strategy: RoutingStrategy) → PriorityRouter[source]: Add a strategy at the end (lowest priority).

insert(index: int, strategy: RoutingStrategy) → PriorityRouter[source]: Insert a strategy at a specific position.

prepend(strategy: RoutingStrategy) → PriorityRouter[source]: Add a strategy at the beginning (highest priority).

class aw.routing.RoutingStrategy(*args, **kwargs)[source]

Protocol for routing strategies.

A routing strategy takes a context and returns a result or None.

aw.routing.detect_explicit(ctx: ExtensionContext) → str | None[source]

Use explicit extension if provided.

>>> ctx = ExtensionContext('', explicit_extension='pdf')
>>> detect_explicit(ctx)
'.pdf'
>>> ctx = ExtensionContext('', explicit_extension='.json')
>>> detect_explicit(ctx)
'.json'

aw.routing.detect_from_url(ctx: ExtensionContext) → str | None[source]

Extract extension from URL path.

>>> ctx = ExtensionContext('https://example.com/file.pdf')
>>> detect_from_url(ctx)
'.pdf'
>>> ctx = ExtensionContext('https://example.com/file')
>>> detect_from_url(ctx) is None
True

aw.routing.make_content_type_detector(content_type_map: dict | None = None) → Callable[[ExtensionContext], str | None][source]

Factory for content-type based detection.

Returns a detector function with the given mapping. Users can create custom detectors with different mappings.

>>> # Use default mapping
>>> detector = make_content_type_detector()
>>> ctx = ExtensionContext('', content_type='application/pdf')
>>> detector(ctx)
'.pdf'
>>>
>>> # Custom mapping
>>> custom_map = {'text/plain': '.log'}
>>> custom_detector = make_content_type_detector(custom_map)
>>> ctx = ExtensionContext('', content_type='text/plain')
>>> custom_detector(ctx)
'.log'

aw.routing.make_magic_bytes_detector(magic_bytes_map: dict | None = None) → Callable[[ExtensionContext], str | None][source]

Factory for magic bytes based detection.

>>> detector = make_magic_bytes_detector()
>>> ctx = ExtensionContext('', content=b'%PDF-1.5...')
>>> detector(ctx)
'.pdf'
>>> ctx = ExtensionContext('', content=b'<html>...')
>>> detector(ctx)
'.html'