# AGENTS.md - Route Function Implementation Guide
**Purpose**: Define conventions, patterns, and best practices for implementing route functions in the ComfyUI MCP server, inspired by the dl_remuxed project architecture.
---
## 🎯 Core Principles
### Single Responsibility Principle (SRP)
Route functions should **only** handle HTTP transport. No business logic, no parsing beyond basic response handling, no retry logic, no polling.
### Dependency Inversion Principle (DIP)
Route functions depend on abstractions (`ComfyAuth`) rather than concrete implementations, allowing flexible authentication strategies.
### Interface Segregation Principle (ISP)
Each route function exposes a minimal, focused interface for a specific API endpoint.
---
## 📐 Route Function Pattern
### Template
```python
from typing import Optional
import httpx
from auth.base import ComfyAuth
from client import get_data as gd
from client.response import ResponseGetData
from utils.logging import log_call
class MyModuleError(Exception):
"""Raised when operations in this module fail."""
def __init__(self, message: str, response: Optional[ResponseGetData] = None):
super().__init__(message)
self.response = response
@log_call(action_name="my_route_function", level_name="route")
async def my_route_function(
auth: ComfyAuth,
required_param: str,
optional_param: str = "default",
*,
session: Optional[httpx.AsyncClient] = None,
) -> ResponseGetData:
"""Brief description of what this route does.
Longer description providing context about the endpoint, what data
it returns, and any important behaviors.
Args:
auth: ComfyAuth instance for base URL and authentication
required_param: Description of required parameter
optional_param: Description of optional parameter (default: "default")
session: Optional HTTPX client for connection pooling
Returns:
ResponseGetData with:
- response: Description of response structure
- status: 200 on success
- is_success: True on success
Raises:
MyModuleError: When the request fails
Example:
>>> auth = NoAuth("http://127.0.0.1:8188")
>>> res = await my_route_function(auth=auth, required_param="value")
>>> print(res.status)
200
"""
# 1. Build URL
url = f"{auth.base_url}/api/endpoint"
# 2. Build params/body
params = {"key": required_param}
body = None # or {"key": "value"} for POST
# 3. Make request via get_data
res = await gd.get_data(
auth=auth,
method="GET", # or "POST", "DELETE", etc.
url=url,
params=params,
body=body,
session=session,
)
# 4. Check success and raise on error
if not res.is_success:
raise MyModuleError(
f"Failed to do X: HTTP {res.status}",
response=res,
)
# 5. Return response
return res
```
---
## 🔧 Required Components
### 1. Authentication Injection
**Always** accept `auth: ComfyAuth` as the first parameter:
```python
async def my_route(auth: ComfyAuth, ...):
url = f"{auth.base_url}/endpoint" # Use auth.base_url
res = await gd.get_data(auth=auth, ...) # Pass auth to get_data
```
**Why?**
- Follows Dependency Inversion Principle
- Enables flexible auth strategies (NoAuth, TokenAuth, future custom auth)
- Centralized header injection in `get_data()`
### 2. Session Pooling
**Always** support optional session parameter:
```python
async def my_route(
auth: ComfyAuth,
*,
session: Optional[httpx.AsyncClient] = None,
):
res = await gd.get_data(auth=auth, ..., session=session)
```
**Why?**
- Enables connection pooling for multiple requests
- Reduces latency with persistent connections
- Caller can manage session lifecycle for batch operations
### 3. Decorator-Based Logging
**Always** use `@log_call` decorator:
```python
from utils.logging import log_call
@log_call(action_name="queue_workflow", level_name="route")
async def queue_workflow(auth: ComfyAuth, ...):
...
```
**Parameters:**
- `action_name`: Human-readable action identifier (defaults to function name)
- `level_name`: Log category (usually "route", "client", or "auth")
- `log_level`: Python log level ("DEBUG", "INFO", "WARNING", "ERROR")
- `log_params`: Enable parameter logging (default: False)
- `sensitive_params`: List of parameter names to redact (e.g., ["url", "api_key", "password"])
**Parameter Sanitization Example:**
```python
@log_call(
action_name="download_model",
level_name="route",
log_params=True,
sensitive_params=["url"] # URL may contain API tokens
)
async def download_model_from_url(url: str, ...):
...
```
**Output:**
```
[abc123] [route] download_model - ENTER params={'url': '[REDACTED]', ...}
[abc123] [route] download_model - SUCCESS (2.345s)
```
**Why?**
- Automatic entry/exit/timing logging without cluttering code
- Correlation ID tracking across layers
- Security: Automatic redaction of sensitive parameters
- Consistent log format across all routes
- Aspect-oriented programming (AOP) pattern
**See Also:** `.github/skills/mcp-logging/SKILL.md` for complete logging documentation
### 4. Custom Exceptions
**Always** define module-specific exceptions:
```python
class WorkflowError(Exception):
"""Raised when workflow operations fail."""
def __init__(self, message: str, response: Optional[ResponseGetData] = None):
super().__init__(message)
self.response = response
```
**Why?**
- Enables fine-grained error handling by callers
- Preserves response data for debugging
- Clear separation between different failure modes
### 5. Standardized Returns
**Always** return `ResponseGetData`:
```python
async def my_route(...) -> ResponseGetData:
res = await gd.get_data(...)
return res # Don't parse or transform here
```
**Why?**
- Consistent interface across all routes
- Caller decides how to parse/validate response
- Follows Interface Segregation Principle
---
## 📦 Module Organization
### File Structure
```
src/routes/
├── __init__.py # Export all route functions
├── AGENTS.md # This file
├── workflow.py # Workflow queue/history routes
├── queue.py # Queue management routes
├── assets.py # Asset retrieval routes
├── models.py # Model discovery routes
└── [future_module].py # Additional endpoint groups
```
### Module Guidelines
1. **One file per logical endpoint group**
- Related operations in same file
- E.g., `workflow.py` contains queue_workflow, get_history, get_prompt_history
2. **Module-level exception class**
- One exception class per module
- E.g., `WorkflowError`, `QueueError`, `AssetError`
3. **Import pattern**
```python
from typing import Optional
import httpx
from auth.base import ComfyAuth
from client import get_data as gd
from client.response import ResponseGetData
from utils.logging import log_call
```
4. **Export in `__init__.py`**
```python
from .workflow import queue_workflow, get_history
from .queue import get_queue, cancel_prompt
__all__ = [
"queue_workflow",
"get_history",
"get_queue",
"cancel_prompt",
]
```
---
## 🚫 Anti-Patterns (What NOT to Do)
### ❌ Business Logic in Routes
**Bad:**
```python
async def queue_workflow(auth, workflow):
# Validate workflow structure
if "nodes" not in workflow:
raise ValueError("Missing nodes")
# Transform workflow
workflow = apply_defaults(workflow)
# Make request
res = await gd.get_data(...)
return res
```
**Good:**
```python
async def queue_workflow(auth, workflow):
# Just transport - validation happens in orchestrator/manager
res = await gd.get_data(...)
return res
```
### ❌ Direct HTTP Calls
**Bad:**
```python
import httpx
async def queue_workflow(auth, workflow):
async with httpx.AsyncClient() as client:
response = await client.post(
f"{auth.base_url}/prompt",
json={"prompt": workflow},
)
return response.json()
```
**Good:**
```python
async def queue_workflow(auth, workflow):
res = await gd.get_data(
auth=auth,
method="POST",
url=f"{auth.base_url}/prompt",
body={"prompt": workflow},
)
return res
```
### ❌ Manual Logging
**Bad:**
```python
import logging
logger = logging.getLogger(__name__)
async def queue_workflow(auth, workflow):
logger.info("Queueing workflow...")
res = await gd.get_data(...)
logger.info("Workflow queued successfully")
return res
```
**Good:**
```python
@log_call(action_name="queue_workflow", level_name="route")
async def queue_workflow(auth, workflow):
res = await gd.get_data(...)
return res
```
### ❌ Retry Logic
**Bad:**
```python
async def queue_workflow(auth, workflow):
for attempt in range(3):
try:
res = await gd.get_data(...)
return res
except Exception:
if attempt == 2:
raise
await asyncio.sleep(1)
```
**Good:**
```python
# Retry logic belongs in orchestrator or client layer, not routes
async def queue_workflow(auth, workflow):
res = await gd.get_data(...)
return res
```
### ❌ Response Parsing
**Bad:**
```python
async def queue_workflow(auth, workflow):
res = await gd.get_data(...)
# Extract and validate response fields
if "prompt_id" not in res.response:
raise ValueError("Missing prompt_id")
return res.response["prompt_id"] # Return specific field
```
**Good:**
```python
async def queue_workflow(auth, workflow):
res = await gd.get_data(...)
return res # Return full ResponseGetData, let caller parse
```
---
## 🔄 Migration Path from Old Code
When refactoring existing code to use routes:
### Before (Old Pattern)
```python
# In comfyui_client.py
class ComfyUIClient:
def _queue_workflow(self, workflow):
response = requests.post(
f"{self.base_url}/prompt",
json={"prompt": workflow},
timeout=30,
)
response.raise_for_status()
return response.json()
```
### After (New Pattern)
**1. Create route function:**
```python
# In src/routes/workflow.py
@log_call(action_name="queue_workflow", level_name="route")
async def queue_workflow(auth: ComfyAuth, workflow: dict, *, session=None):
res = await gd.get_data(
auth=auth,
method="POST",
url=f"{auth.base_url}/prompt",
body={"prompt": workflow},
session=session,
)
if not res.is_success:
raise WorkflowError(f"Failed to queue: HTTP {res.status}", response=res)
return res
```
**2. Update orchestrator to use route:**
```python
# In comfyui_client.py (refactored as orchestrator)
import asyncio
from routes import queue_workflow
class ComfyUIOrchestrator:
def _queue_workflow(self, workflow):
# Bridge sync to async
res = asyncio.run(queue_workflow(
auth=self.auth,
workflow=workflow,
))
return res.response # Parse response here, not in route
```
---
## 🧪 Testing Routes
### Unit Test Pattern
```python
import pytest
from routes.workflow import queue_workflow
from auth.base import NoAuth
@pytest.mark.asyncio
async def test_queue_workflow_success(httpx_mock):
"""Test successful workflow queueing."""
# Arrange
auth = NoAuth("http://127.0.0.1:8188")
workflow = {"3": {"class_type": "KSampler"}}
httpx_mock.add_response(
method="POST",
url="http://127.0.0.1:8188/prompt",
json={"prompt_id": "abc-123", "number": 1},
status_code=200,
)
# Act
res = await queue_workflow(auth=auth, workflow=workflow)
# Assert
assert res.is_success
assert res.status == 200
assert res.response["prompt_id"] == "abc-123"
@pytest.mark.asyncio
async def test_queue_workflow_error(httpx_mock):
"""Test workflow queueing failure."""
# Arrange
auth = NoAuth("http://127.0.0.1:8188")
workflow = {"3": {"class_type": "KSampler"}}
httpx_mock.add_response(
method="POST",
url="http://127.0.0.1:8188/prompt",
status_code=500,
)
# Act & Assert
with pytest.raises(WorkflowError) as exc_info:
await queue_workflow(auth=auth, workflow=workflow)
assert "Failed to queue" in str(exc_info.value)
assert exc_info.value.response.status == 500
```
---
## 📚 Related Documentation
### Implementation References
- **Auth Module**: [src/auth/base.py](../auth/base.py) - Authentication abstractions
- **Client Module**: [src/client/get_data.py](../client/get_data.py) - HTTP client implementation
- **Logging Utilities**: [src/utils/logging.py](../utils/logging.py) - Log decorator implementation
- **Reference Project**: GitHub/dl_remuxed/src/routes/ - Original pattern inspiration
### Related Skills
- **[MCP Logging](../../.github/skills/mcp-logging/SKILL.md)** - Correlation ID tracking and structured logging patterns
- **[FastMCP v3](../../.github/skills/fastmcp-v3/SKILL.md)** - FastMCP patterns for context managers
---
## 🔮 Future Enhancements
1. **Retry Decorator**: Add `@run_with_retry()` decorator to client layer
2. **Response Caching**: Implement cached transport for GET requests
3. **Rate Limiting**: Add rate limiter for external API calls
4. **Request Context**: Introduce RouteContext for aggregating config (timeout, verify, cache)
5. **Async Batching**: Helper for parallel route calls with session pooling
---
## ✅ Checklist for New Routes
When adding a new route function:
- [ ] Uses `@log_call` decorator
- [ ] Accepts `auth: ComfyAuth` as first parameter
- [ ] Supports optional `session: Optional[httpx.AsyncClient]` parameter
- [ ] Returns `ResponseGetData`
- [ ] Raises module-specific exception on error
- [ ] Has comprehensive docstring with Args/Returns/Raises/Example
- [ ] No business logic (validation, parsing, transformation)
- [ ] No manual logging calls
- [ ] No direct HTTPX calls (uses `get_data()`)
- [ ] Exported in `__init__.py`
- [ ] Has unit tests with mocked HTTP responses
---
**Last Updated**: January 23, 2026
**Maintainer**: Refactoring Team
**Status**: Active Development
