Agent Coordination Quick Reference

def _create_agent_tool(agent_name: str, agent, tool_name: str, description: str):
    @tool(tool_name, description=description)
    def invoke_agent(task_description: str) -> str:
        config = {"run_name": agent_name, "tags": [agent_name]}
        result = agent.invoke(
            {"messages": [{"role": "user", "content": task_description}]},
            config=config
        )
        return result.get("messages", [])[-1].content
    return invoke_agent

is_ready_for_handoff = (
    bool(extracted_identifiers) and
    bool(dataset_ids) and
    bool(workspace_metadata_keys)
)
# → Status transitions to HANDOFF_READY

permission_request = f"""
Disease validation failed: {coverage:.1f}% (required: 50%)

I can attempt automatic enrichment to extract disease from publication abstracts.
- Estimated time: 20 seconds
- Cost: ~$0.03 in LLM tokens

Should I proceed with automatic enrichment?"""

# Write to workspace
content = MetadataContent(
    identifier="sra_PRJNA123_samples",
    data={"samples": sample_list},
    source="research_agent"
)
workspace_service.write_content(content, ContentType.METADATA)

queue.update_status(
    entry_id,
    PublicationStatus.HANDOFF_READY,
    handoff_status=HandoffStatus.READY_FOR_METADATA,
    workspace_metadata_keys=["sra_PRJNA123_samples"]
)

with queue_file_lock(thread_lock, lock_path):
    atomic_write_jsonl(queue_path, entries, serializer)

PENDING → EXTRACTING → METADATA_EXTRACTED → METADATA_ENRICHED → HANDOFF_READY → COMPLETED
                                                                          ↓
                                                                      FAILED

if workspace_key not in metadata_store:
    # Try workspace
    ws_data = workspace_service.read_content(workspace_key)
    if not ws_data:
        return "❌ Not found. Use get_content_from_workspace() to list available files"

if disease_coverage < min_disease_coverage:
    if strict_disease_validation:
        return f"❌ Failed: {coverage:.1f}% (need {min_disease_coverage*100:.1f}%)"
    else:
        logger.warning(f"Low coverage: {coverage:.1f}%")
        # Continue with warning

# In supervisor.py:120-130
if "ADMIN SUPERUSER" in user_message:
    # Bypass ALL confirmations
    # Execute immediately
    # No permission checks