File size: 4,131 Bytes
5dc5419 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """SyntheticOutputTool - Generate structured synthetic output for Stack 2.9"""
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
from .base import BaseTool, ToolResult
from .registry import tool_registry
class SyntheticOutputTool(BaseTool):
"""Generate structured synthetic output."""
name = "synthetic_output"
description = "Generate structured synthetic output in various formats"
input_schema = {
"type": "object",
"properties": {
"output_type": {
"type": "string",
"enum": ["json", "markdown", "text", "html"],
"description": "Output format"
},
"content": {"type": "string", "description": "Content to format"},
"metadata": {"type": "object", "description": "Optional metadata"}
},
"required": ["output_type", "content"]
}
async def execute(self, output_type: str, content: str, metadata: Optional[Dict] = None) -> ToolResult:
"""Generate output."""
result = {
"type": output_type,
"content": content,
"generated_at": datetime.now().isoformat()
}
if metadata:
result["metadata"] = metadata
if output_type == "json":
formatted = json.dumps(result, indent=2)
elif output_type == "markdown":
formatted = self._to_markdown(result)
elif output_type == "html":
formatted = self._to_html(result)
else:
formatted = content
return ToolResult(success=True, data={
"type": output_type,
"formatted": formatted,
"raw": result
})
def _to_markdown(self, data: Dict) -> str:
"""Convert to markdown."""
lines = [f"# Synthetic Output", f"**Type:** {data.get('type')}", f"**Generated:** {data.get('generated_at')}", ""]
if data.get("metadata"):
lines.append("## Metadata")
for k, v in data["metadata"].items():
lines.append(f"- **{k}:** {v}")
lines.append("")
lines.append("## Content")
lines.append(data.get("content", ""))
return '\n'.join(lines)
def _to_html(self, data: Dict) -> str:
"""Convert to HTML."""
meta = ""
if data.get("metadata"):
meta = "<ul>" + "".join(f"<li><strong>{k}:</strong> {v}</li>" for k, v in data["metadata"].items()) + "</ul>"
return f"""
<div class="synthetic-output">
<h2>Synthetic Output</h2>
<p><strong>Type:</strong> {data.get('type')}</p>
<p><strong>Generated:</strong> {data.get('generated_at')}</p>
{meta}
<div class="content">
<pre>{data.get('content', '')}</pre>
</div>
</div>
"""
class StructuredDataTool(BaseTool):
"""Convert unstructured data to structured format."""
name = "structure_data"
description = "Convert unstructured data to structured format"
input_schema = {
"type": "object",
"properties": {
"data": {"type": "string", "description": "Data to structure"},
"schema": {"type": "object", "description": "Target schema"},
"format": {"type": "string", "enum": ["json", "csv"], "default": "json"}
},
"required": ["data"]
}
async def execute(self, data: str, schema: Optional[Dict] = None, format: str = "json") -> ToolResult:
"""Structure data."""
# Simple JSON detection
try:
parsed = json.loads(data)
return ToolResult(success=True, data={
"parsed": parsed,
"format": "json",
"structured": True
})
except json.JSONDecodeError:
pass
# Treat as plain text
return ToolResult(success=True, data={
"data": data,
"format": "text",
"structured": False,
"note": "Could not auto-structure, returned as text"
})
# Register tools
tool_registry.register(SyntheticOutputTool())
tool_registry.register(StructuredDataTool())
|