get_poems_for_enrichment
Retrieve poems requiring theme analysis to enhance their metadata classification. Provides poems with minimal tags for systematic theme identification and enrichment.
Instructions
Get batch of poems needing theme enrichment for agent analysis.
Returns poems with minimal or no tags for YOU (the agent) to analyze. YOU suggest which themes apply to each poem.
Args: poem_ids: List of poem IDs (None = all untagged/lightly-tagged poems) max_poems: Maximum poems to return (default 50)
Returns: Dictionary with: - poems: List of poem data (id, title, content, current_tags) - available_themes: Theme options with descriptions - instructions: Batch analysis guidance
Example workflow: ``` # 1. Get poems needing enrichment data = await get_poems_for_enrichment(max_poems=10)
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| poem_ids | No | ||
| max_poems | No |
Implementation Reference
- Core implementation of get_poems_for_enrichment. Selects untagged or lightly-tagged poems from catalog, loads and truncates their content, prepares list of available themes with brief descriptions, and returns structured data including analysis instructions for batch agent processing.async def get_poems_for_enrichment( poem_ids: Optional[List[str]] = None, max_poems: int = 50, ) -> dict: """Get list of poems needing theme enrichment for agent analysis. Returns poems with minimal or no tags for the agent to analyze and suggest themes. The agent can analyze multiple poems and suggest which themes to apply. Args: poem_ids: List of poem IDs to include (None = all untagged poems) max_poems: Maximum poems to return (default 50) Returns: Dictionary with: - poems: List of poem data (id, title, content, current_tags, state) - available_themes: Theme options with descriptions - total_count: Total poems returned - instructions: Guidance for batch analysis Example: >>> result = await get_poems_for_enrichment(max_poems=10) >>> # Agent analyzes result['poems'] against result['available_themes'] >>> # Agent suggests themes for each poem >>> # User applies tags with link_poem_to_nexus() """ if _catalog is None or _nexus_registry is None: raise RuntimeError("Enrichment tools not initialized.") # Determine which poems to return if poem_ids is None: # Default: all poems with no tags or very few tags poems_to_return = [ poem for poem in _catalog.index.all_poems if not poem.tags or len(poem.tags) < 2 ] else: # Specific poem IDs poems_to_return = [] for pid in poem_ids: poem = _catalog.index.get_poem(pid) if poem: poems_to_return.append(poem) # Limit to max_poems poems_to_return = poems_to_return[:max_poems] # Format poems data poems_data = [] for poem in poems_to_return: # Load content if needed content = poem.content if not content: try: poem_path = Path(poem.file_path) full_content = poem_path.read_text(encoding='utf-8') if '---' in full_content: parts = full_content.split('---', 2) if len(parts) >= 3: content = parts[2].strip() else: content = full_content else: content = full_content except Exception: content = "[Content unavailable]" poems_data.append({ "id": poem.id, "title": poem.title, "content": content[:500] + "..." if len(content) > 500 else content, # Truncate for efficiency "current_tags": poem.tags or [], "state": poem.state, }) # Format available themes themes_data = [] for nexus in _nexus_registry.themes: desc_lines = nexus.description.split('\n') brief_desc = [] for line in desc_lines: if line.strip() and not line.startswith('#'): brief_desc.append(line.strip()) if len(brief_desc) >= 2: break brief_text = ' '.join(brief_desc)[:150] themes_data.append({ "name": nexus.name, "canonical_tag": nexus.canonical_tag, "description": brief_text, }) instructions = """Analyze these poems and suggest themes for each. For each poem, provide 1-3 most relevant themes with: - **name**: Theme name (exactly as listed) - **canonical_tag**: Tag to use - **confidence**: Float 0.0-1.0 - **evidence**: Brief reasoning After analysis, user can apply tags with link_poem_to_nexus(poem_id, nexus_name, "theme")""" logger.info(f"Prepared {len(poems_data)} poems for batch enrichment") return { "success": True, "poems": poems_data, "available_themes": themes_data, "total_count": len(poems_data), "instructions": instructions, }
- src/poetry_mcp/server.py:398-429 (registration)MCP server registration of the tool using @mcp.tool() decorator. Provides the official tool interface and docstring for MCP clients/agents, delegating execution to the underlying _get_poems_for_enrichment imported from enrichment_tools.py.@mcp.tool() async def get_poems_for_enrichment( poem_ids: Optional[List[str]] = None, max_poems: int = 50, ) -> dict: """ Get batch of poems needing theme enrichment for agent analysis. Returns poems with minimal or no tags for YOU (the agent) to analyze. YOU suggest which themes apply to each poem. Args: poem_ids: List of poem IDs (None = all untagged/lightly-tagged poems) max_poems: Maximum poems to return (default 50) Returns: Dictionary with: - poems: List of poem data (id, title, content, current_tags) - available_themes: Theme options with descriptions - instructions: Batch analysis guidance Example workflow: ``` # 1. Get poems needing enrichment data = await get_poems_for_enrichment(max_poems=10) # 2. YOU analyze data['poems'] against data['available_themes'] # 3. YOU suggest 1-3 themes for each poem with confidence scores # 4. User applies high-confidence tags with link_poem_to_nexus() ``` """ return await _get_poems_for_enrichment(poem_ids, max_poems)
- src/poetry_mcp/server.py:23-32 (registration)Import of the underlying handler from enrichment_tools.py with alias _get_poems_for_enrichment, enabling the server.py wrapper to delegate tool execution.from .tools.enrichment_tools import ( initialize_enrichment_tools, get_all_nexuses as _get_all_nexuses, link_poem_to_nexus as _link_poem_to_nexus, find_nexuses_for_poem as _find_nexuses_for_poem, get_poems_for_enrichment as _get_poems_for_enrichment, sync_nexus_tags as _sync_nexus_tags, move_poem_to_state as _move_poem_to_state, grade_poem_quality as _grade_poem_quality, )