feat: add capability-based task routing#3
feat: add capability-based task routing#3zakiscoding wants to merge 1 commit intojoshualamerton:mainfrom
Conversation
Closes joshualamerton#1 Adds Task and Agent classes with capability metadata. TaskRouter now matches tasks to agents by required capabilities - claim_task() iterates the queue and returns the first task the agent can handle. Falls back to FIFO for unregistered agents (backward compatible).
There was a problem hiding this comment.
Pull request overview
Implements capability-based task routing in TaskMesh by introducing first-class Task/Agent models and updating the router to match queued tasks to agents based on required capabilities.
Changes:
- Added
TaskandAgentclasses and agent registration toTaskRouter. - Updated
TaskRouter.submit_task()to accept either a string task name or aTaskobject (with optional capability requirements). - Updated the demo to register agents and submit capability-scoped tasks.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
core/task_router.py |
Introduces Task/Agent and capability-aware claim_task() routing logic. |
examples/demo.py |
Demonstrates registering agents with capabilities and claiming capability-matched tasks. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
You can also share your feedback on Copilot code review. Take the survey.
| def __repr__(self): | ||
| caps = ", ".join(self.required_capabilities) if self.required_capabilities else "any" | ||
| return f"Task({self.name\!r}, requires={caps})" |
There was a problem hiding this comment.
Task.__repr__ uses {self.name\!r} (and similar in Agent.__repr__), which is invalid f-string syntax and will raise a SyntaxError. Remove the backslash and use the standard repr conversion {self.name!r} so the module can import/run.
| def __repr__(self): | ||
| return f"Agent({self.name\!r}, capabilities={sorted(self.capabilities)})" |
There was a problem hiding this comment.
Agent.__repr__ uses {self.name\!r}, which is invalid f-string syntax and will raise a SyntaxError. Remove the backslash and use {self.name!r}.
| def __init__(self, name: str, required_capabilities: list[str] | None = None): | ||
| self.name = name | ||
| self.required_capabilities = required_capabilities or [] |
There was a problem hiding this comment.
README advertises Python 3.9+, but these type hints use PEP 604 unions (list[str] | None), which require Python 3.10+. To keep 3.9 compatibility, switch to Optional[...] / Union[...] (or bump the documented minimum Python version).
| self.tasks.append(task) | ||
|
|
||
| def claim_task(self, agent_name): | ||
| def claim_task(self, agent_name: str) -> dict | None: |
There was a problem hiding this comment.
This return annotation uses dict | None, which requires Python 3.10+. If TaskMesh targets Python 3.9+ (per README), use Optional[dict[...]] (and import Optional from typing) or update the documented Python requirement.
|
|
||
| Iterates through the task queue in order and returns the first task | ||
| whose required capabilities are all present in the agent's capability set. | ||
| Falls back to the first available task if the agent has no capabilities registered. |
There was a problem hiding this comment.
Docstring says routing “falls back to the first available task if the agent has no capabilities registered,” but the implementation only falls back when the agent is not registered (agent is None). For registered agents with an empty capability set, the router will only match tasks with no required capabilities. Please update the docstring (or adjust logic) so behavior and docs align.
| Falls back to the first available task if the agent has no capabilities registered. | |
| Falls back to the first available task if the agent is not registered. |
|
Hi @joshualamerton! This PR is ready for review and merge. It implements capability-based task routing with |
Summary
Implements capability-based task routing in TaskMesh.
New classes
Task(name, required_capabilities)— task with optional capability requirementsAgent(name, capabilities)— agent with a defined capability setTaskRouter.register_agent(agent)— register an agentTaskRouter.claim_task()now matches tasks to agents by capabilityHow it works
claim_task()iterates the queue and returns the first task the agent is capable of handling (all required capabilities present). Falls back to FIFO for unregistered agents (backward compatible).Example
Closes #1