feat(steering): allow steering on AfterModelCallEvents

[dbschmigelski]

Description

Expands the steering system to support model response steering via AfterModelCallEvent, addressing a key limitation where steering could only influence tool selection and execution.

Previously, steering handlers could only intercept tool calls via steer() (now steer_before_tool()). This meant there was no way to:

• Validate or retry model responses that don't meet quality criteria
• Ensure required tools are used before an agent completes
• Guide conversation flow based on model output

This PR adds steer_after_model() which is called after each model response, enabling handlers to:

• Proceed: Accept the response as-is
• Guide: Discard the response and retry with guidance injected into the conversation

A practical example from the issue: ensuring an agent sends a required confirmation message or uses a mandatory tool before completing a workflow. The new steer_after_model() hook intercepts end_turn responses and can force retries with guidance until requirements are met.

API Changes

New method on SteeringHandler:

async def steer_after_model(
    self, agent: Agent, message: Message, stop_reason: StopReason, **kwargs
) -> ModelSteeringAction:
    """Override to implement custom model steering logic."""
    return Proceed(reason="Default: accept response")

New type aliases for type safety:

• ToolSteeringAction = Proceed | Guide | Interrupt - for steer_before_tool()
• ModelSteeringAction = Proceed | Guide - for steer_after_model()
  • No Interrupt until [FEATURE] Support Interrupt from any HookEvent #1165 since we can only interrupt on tool steering. I chose to break symmetry in favor of "compile" time checks rather than a runtime exception or no op.

Renamed with deprecation:

• steer() → steer_before_tool() (old method emits DeprecationWarning)

Example: Enforcing Tool Usage

class ForceToolUsageHandler(SteeringHandler):
    def __init__(self, required_tool: str):
        super().__init__()
        self.required_tool = required_tool

    async def steer_after_model(self, agent, message, stop_reason, **kwargs):
        if stop_reason != "end_turn":
            return Proceed(reason="Model still processing")

        # Check if required tool was used
        for block in message.get("content", []):
            if "toolUse" in block and block["toolUse"].get("name") == self.required_tool:
                return Proceed(reason="Required tool was used")

        # Force tool usage
        return Guide(reason=f"You MUST use the {self.required_tool} tool before completing.")

# Agent will retry until it uses the required tool
agent = Agent(tools=[log_activity], hooks=[ForceToolUsageHandler("log_activity")])

Related Issues

This PR move the needle for #1386.

Type of Change

New feature

Testing

• Reorganized test_tool_steering.py for existing tool steering tests

• Updated unit tests in test_handler.py for new method signatures

• I ran hatch run prepare

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[dbschmigelski]

We are failing with "Potential API changes detected (review if actually breaking)" this is true, but we are in experimental so this should not strictly be a blocker.

[strands-agent]

🎯 Review - Steering on AfterModelCallEvents

This is an excellent feature that addresses a key limitation in the steering system! The ability to steer after model responses opens up many valuable use cases like quality checks, required tool enforcement, and conversation flow control.

What I Really Like ✅

1. Clear API Design: The separation of steer_before_tool() and steer_after_model() with distinct return types (ToolSteeringAction vs ModelSteeringAction) provides compile-time safety. Smart choice!

2. Deprecation Handling: Renaming steer() → steer_before_tool() with a proper deprecation warning shows good API stewardship.

3. Comprehensive Testing: 204 lines of new integration tests demonstrating real-world scenarios (enforcing tool usage, response validation).

4. Thorough Documentation: The PR description explains the motivation, API changes, and includes clear examples. Very helpful!

5. Already Approved: @zastrowm approved this, which is great!

Addressing the CI Failure ⚠️

I see the check-api failure with "Potential API changes detected (review if actually breaking)". You mentioned this is expected since the feature is in experimental.

My take: Since this is in strands.experimental.steering, API changes are acceptable and expected. The deprecation warning for steer() provides backward compatibility for the transition period.

For maintainers: This failure seems like it can be safely overridden given:

• Feature is in experimental namespace
• Backward compatibility maintained via deprecation
• API changes are additive (new method) with minimal breaking changes

Minor Suggestions 💡

1. Documentation Enhancement

The example in the PR description is great! Consider adding inline comments for clarity:

class ForceToolUsageHandler(SteeringHandler):
    def __init__(self, required_tool: str):
        super().__init__()
        self.required_tool = required_tool

    async def steer_after_model(self, agent, message, stop_reason, **kwargs):
        # Only intercept when agent is trying to end the conversation
        if stop_reason != "end_turn":
            return Proceed(reason="Model still processing")

        # Check if required tool was used in the conversation history
        for block in message.get("content", []):
            if "toolUse" in block and block["toolUse"].get("name") == self.required_tool:
                return Proceed(reason="Required tool was used")

        # Force retry with guidance - agent will see this as a system message
        return Guide(reason=f"You MUST use the {self.required_tool} tool before completing.")

2. Type Hint Enhancement

In handler.py, consider adding more specific return type documentation:

async def steer_after_model(
    self, agent: Agent, message: Message, stop_reason: StopReason, **kwargs
) -> ModelSteeringAction:
    """Override to implement custom model steering logic.
    
    Called after each model response to validate or guide the conversation.
    
    Args:
        agent: The agent instance
        message: The model's response message
        stop_reason: Why the model stopped (end_turn, tool_use, max_tokens, etc.)
        **kwargs: Additional context
    
    Returns:
        Proceed: Accept the response as-is
        Guide: Reject response and retry with guidance injected into conversation
    
    Note:
        Interrupt is not supported for model steering (only for tool steering).
        Use Guide with specific instructions to achieve similar behavior.
    """
    return Proceed(reason="Default: accept response")

3. Integration Test Suggestion

The tests are comprehensive! One additional test case that might be valuable:

async def test_steering_prevents_infinite_retry_loop():
    """Ensure steering doesn't cause infinite loops if Guide is always returned."""
    # Test that there's a max retry limit or timeout
    # This protects against buggy steering handlers

Questions for Discussion 🤔

1. Retry Limits: Is there a maximum number of times Guide can force a retry? Should there be a circuit breaker to prevent infinite loops from buggy handlers?

2. Guide Message Visibility: When Guide injects guidance into the conversation, is this visible in the agent's message history or only used for the retry?

3. Future Interrupt Support: You mentioned [FEATURE] Support Interrupt from any HookEvent #1165 for interrupting on model responses. What's the timeline for that feature?

4. Performance Impact: Does model steering add noticeable latency since it checks after every model call?

Real-World Use Cases 🌟

This feature enables some really powerful patterns:

• Quality gates: Ensure responses meet length/format requirements
• Compliance checks: Validate that sensitive topics are handled appropriately
• Workflow enforcement: Require specific tools before completion (your example)
• Multi-agent coordination: Guide agents to hand off to other agents

Overall Assessment

This is a well-designed, thoroughly tested feature that significantly expands the steering system's capabilities. The experimental status is appropriate, and the API design is clean.

✅ Recommendation: Merge once maintainers are comfortable with the API changes in experimental.

Great work, @dbschmigelski! This will be very useful for the community. 🎉

🦆

---

🤖 This is an experimental AI agent response from the Strands team, powered by Strands Agents. We're exploring how AI agents can help with community support and development. Your feedback helps us improve! If you'd prefer human assistance, please let us know.