Fix/277 runner lifecycling

[CasperGN]

Description

This also carries a fix to the quickstarter tests.

Moving lifecycling of DurableAgent into AgentRunner for a more streamlined instantiation.
The AgentRunner will now handle running agent.start() in

• run
• serve
• subscribe
• register_routes

It will catch the RuntimeError thrown by the agent if already running and pass.

AgentRunner().shutdown() now overrides and takes the agent: Any parameter to call agent.stop() moving the lifecycle of the agent into the runner:

Before:

agent = DurableAgent()
agent.start()
runner = AgentRunner()
try:
    runner.serve(agent, port=8001)
finally:
    runner.shutdown()
    agent.stop()

Now:

agent = DurableAgent()
runner = AgentRunner()
try:
    runner.serve(agent, port=8001)
finally:
    runner.shutdown(agent)

Issue reference

We strive to have all PR being opened based on an issue, where the problem or feature have been discussed prior to implementation.

Please reference the issue this PR will close: Closes #277

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Created/updated tests
• Tested this change against all the quickstarts
• Extended the documentation
  • Created the dapr/docs PR: chore: align docs with https://github.com/dapr/dapr-agents/pull/283 docs#4955

No tests needed updates as the agents are never passed to an AgentRunner.
tests/integration/quickstarts/test_03_durable_agent_tool_call.py has a dependency on arize-phoenix-otel - @Cyb3rWard0g is this intentional or a leftover? See here

[bibryam]

@CasperGN this simplifies runner usage significantly.

IMO shutdown could be simplified tiny bit more.
Have you considered these 2 scenarios around shutdown:

# Create multiple agents
agent1 = DurableAgent(name="Agent1", ...)
agent2 = DurableAgent(name="Agent2", ...)
agent3 = DurableAgent(name="Agent3", ...)

# Create a runner and serve/subscribe multiple agents
runner = AgentRunner()

# Serve multiple agents
runner.serve(agent1, port=8001)
runner.subscribe(agent2)
runner.serve(agent3, port=8002)  # agent3 uses serve instead of subscribe

# Scenario 1: Shutdown a specific agent
# This should only stop agent2, leaving agent1 and agent3 running
runner.shutdown(agent2)

# Scenario 2: Shutdown all agents
# This should stop all remaining agents (agent1 and agent3)
runner.shutdown()

# Expected behavior:
# - runner.shutdown(agent) -> stops only that specific agent
# - runner.shutdown() -> stops all agents managed by the runner

[CasperGN]

@bibryam actually now that I think more about this, the runner.shutdown() is problematic since it only takes a single agent entry and you're absolutely right, we could have multiple agents.

What about something like this:

# then AgentRunner

class AgentRunner()...

  def __init__():
    self.managed_agents = []
    ..

  def run(): # and the other func's that calls agent.start()
    agent.start()
    self.managed_agents.append(agent)

  def shutdown(agent: Optional[Any]), ..):
    if agent:
      if agent in self.managed_agents:
        agent.stop()
        self.managed_agents.remove(agent)
        return
    # continue with full shutdown

Thoughts?

[Cyb3rWard0g]

LGTM! Thank you @CasperGN !