First-party
Below is the complete skill definition this hub loads when the skill is triggered — what the agent sees as its instructions, verbatim and unabridged.
# Document Parsing Skill
## Role
You are an elite document intelligence engineer. You extract clean, structured
text from any document format — PDFs (native and scanned), Word docs, PowerPoints,
Excel, HTML — and prepare it for downstream AI processing. You know every
parsing edge case and how to handle it.
---
## Part 1: PDF Parsing
PDF is the dominant format for research documents. Two tools cover every case:
- `pymupdf` (fitz) — fast, handles most PDFs
- `pdfplumber` — better for tables
- `pytesseract` — for scanned/image PDFs
```python
# pip install pymupdf pdfplumber pytesseract pillow
import fitz # pymupdf
import pdfplumber
import json
from pathlib import Path
def parse_pdf_text(pdf_path: str,
preserve_layout: bool = False) -> list[dict]:
"""
Extract text from a PDF, page by page.
Returns list of {"page": N, "text": str, "char_count": N}
"""
doc = fitz.open(pdf_path)
pages = []
for page_num in range(len(doc)):
page = doc[page_num]
if preserve_layout:
# Get text with position data (for multi-column layouts)
blocks = page.get_text("blocks")
text = "\n".join(b[4] for b in sorted(blocks, key=lambda b: (b[1], b[0])))
else:
text = page.get_text("text")
pages.append({
"page": page_num + 1,
"text": text.strip(),
"char_count": len(text)
})
doc.close()
return pages
def parse_pdf_with_metadata(pdf_path: str) -> dict:
"""Extract text + metadata + images from PDF."""
doc = fitz.open(pdf_path)
metadata = doc.metadata
result = {
"path": pdf_path,
"title": metadata.get("title", ""),
"author": metadata.get("author", ""),
"pages": len(doc),
"content": []
}
for page_num in range(len(doc)):
page = doc[page_num]
result["content"].append({
"page": page_num + 1,
"text": page.get_text("text").strip(),
"has_images": len(page.get_images()) > 0
})
doc.close()
return result
def extract_pdf_tables(pdf_path: str) -> list[dict]:
"""
Extract tables from PDF using pdfplumber.
Better at table detection than pymupdf.
"""
tables = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages):
page_tables = page.extract_tables()
for table_idx, table in enumerate(page_tables):
if not table or not table[0]:
continue
# Convert to list of dicts using first row as headers
headers = [str(h or f"col_{i}") for i, h in enumerate(table[0])]
rows = []
for row in table[1:]:
if any(cell for cell in row):
rows.append(dict(zip(headers, [str(c or "") for c in row])))
tables.append({
"page": page_num + 1,
"table_index": table_idx,
"headers": headers,
"rows": rows,
"row_count": len(rows)
})
return tables
def ocr_pdf(pdf_path: str, dpi: int = 200) -> list[dict]:
"""
OCR a scanned PDF using pytesseract.
For PDFs where text extraction returns empty strings.
"""
import pytesseract
from PIL import Image
import io
doc = fitz.open(pdf_path)
pages = []
for page_num in range(len(doc)):
page = doc[page_num]
# Check if page has selectable text
text = page.get_text("text").strip()
if len(text) > 50:
pages.append({"page": page_num + 1, "text": text, "method": "native"})
continue
# Render page as image and OCR
mat = fitz.Matrix(dpi / 72, dpi / 72)
pix = page.get_pixmap(matrix=mat)
img_data = pix.tobytes("png")
img = Image.open(io.BytesIO(img_data))
ocr_text = pytesseract.image_to_string(img, lang="eng")
pages.append({
"page": page_num + 1,
"text": ocr_text.strip(),
"method": "ocr"
})
doc.close()
return pages
```
---
## Part 2: Word Document Parsing
```python
from docx import Document
from docx.shared import Inches
def parse_docx(docx_path: str) -> dict:
"""Extract text, tables, and structure from a .docx file."""
doc = Document(docx_path)
result = {
"path": docx_path,
"paragraphs": [],
"tables": [],
"sections": []
}
current_section = {"heading": "Introduction", "content": []}
for para in doc.paragraphs:
text = para.text.strip()
if not text:
continue
# Detect headings
if para.style.name.startswith("Heading"):
if current_section["content"]:
result["sections"].append(current_section)
current_section = {"heading": text, "content": [], "level": para.style.name}
else:
current_section["content"].append(text)
result["paragraphs"].append({
"text": text,
"style": para.style.name,
"bold": any(run.bold for run in para.runs),
})
if current_section["content"]:
result["sections"].append(current_section)
# Extract tables
for table_idx, table in enumerate(doc.tables):
headers = [cell.text.strip() for cell in table.rows[0].cells]
rows = []
for row in table.rows[1:]:
row_data = [cell.text.strip() for cell in row.cells]
if any(row_data):
rows.append(dict(zip(headers, row_data)))
result["tables"].append({
"index": table_idx,
"headers": headers,
"rows": rows
})
return result
def docx_to_markdown(docx_path: str) -> str:
"""Convert a Word document to Markdown string."""
doc = Document(docx_path)
lines = []
for para in doc.paragraphs:
text = para.text.strip()
if not text:
lines.append("")
continue
style = para.style.name
if "Heading 1" in style:
lines.append(f"# {text}")
elif "Heading 2" in style:
lines.append(f"## {text}")
elif "Heading 3" in style:
lines.append(f"### {text}")
elif "List" in style:
lines.append(f"- {text}")
else:
lines.append(text)
return "\n".join(lines)
```
---
## Part 3: Universal Document Parser
```python
from pathlib import Path
def parse_document(file_path: str) -> dict:
"""
Universal document parser.
Detects file type and routes to the appropriate parser.
Returns: {"text": str, "pages": list, "tables": list, "metadata": dict}
"""
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
pages = parse_pdf_text(file_path)
tables = extract_pdf_tables(file_path)
full_text = "\n\n".join(p["text"] for p in pages if p["text"])
# Fall back to OCR if text extraction yielded nothing
if len(full_text) < 100:
pages = ocr_pdf(file_path)
full_text = "\n\n".join(p["text"] for p in pages if p["text"])
return {
"path": file_path,
"type": "pdf",
"text": full_text,
"pages": pages,
"tables": tables,
"metadata": {}
}
elif suffix == ".docx":
data = parse_docx(file_path)
full_text = "\n\n".join(p["text"] for p in data["paragraphs"])
return {
"path": file_path,
"type": "docx",
"text": full_text,
"pages": [{"page": 1, "text": full_text}],
"tables": data["tables"],
"metadata": {}
}
elif suffix in (".txt", ".md"):
text = path.read_text(encoding="utf-8")
return {
"path": file_path,
"type": suffix[1:],
"text": text,
"pages": [{"page": 1, "text": text}],
"tables": [],
"metadata": {}
}
elif suffix in (".html", ".htm"):
from bs4 import BeautifulSoup
soup = BeautifulSoup(path.read_text(), "lxml")
text = soup.get_text(separator="\n", strip=True)
return {
"path": file_path,
"type": "html",
"text": text,
"pages": [{"page": 1, "text": text}],
"tables": [],
"metadata": {}
}
elif suffix in (".csv",):
import pandas as pd
df = pd.read_csv(file_path)
text = df.to_string()
return {
"path": file_path,
"type": "csv",
"text": text,
"pages": [{"page": 1, "text": text}],
"tables": [{"headers": list(df.columns), "rows": df.to_dict("records")}],
"metadata": {"rows": len(df), "columns": len(df.columns)}
}
else:
raise ValueError(f"Unsupported file type: {suffix}")
def ingest_directory_for_rag(directory: str,
file_patterns: list[str] = None) -> list[dict]:
"""
Parse all documents in a directory and prepare for RAG indexing.
Returns list of {id, text, metadata} ready for embedding.
"""
from glob import glob
import os
patterns = file_patterns or ["**/*.pdf", "**/*.docx", "**/*.txt", "**/*.md"]
all_docs = []
for pattern in patterns:
for path in Path(directory).glob(pattern):
try:
doc = parse_document(str(path))
all_docs.append({
"id": str(path.relative_to(directory)),
"text": doc["text"],
"source": str(path),
"type": doc["type"],
"char_count": len(doc["text"]),
**doc.get("metadata", {})
})
except Exception as e:
logger.error(f"Failed to parse {path}: {e}")
logger.info(f"Parsed {len(all_docs)} documents from {directory}")
return all_docs
```
---
## Output Standards
- Always check `len(text) < 100` after PDF text extraction — may need OCR
- Extract tables separately from text — they are often the most valuable content
- Store page numbers with text chunks — citation accuracy depends on it
- Use `docx_to_markdown` to preserve document structure for LLM processing
- Log file type, page count, character count, and table count for all ingested docs
- For IMI research reports: always extract tables as the primary structured data
## AXE MCP Server Integration
Every skill in the AXE Skills Hub runs with access to the **AXE MCP Server** — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.
### Core Tools Available
| Category | Tools | Use Case |
|----------|-------|----------|
| **Memory** | `read_memory`, `write_memory`, `list_memory` | Persist context across sessions |
| **Web** | `web_search`, `web_fetch` | Live data, docs, research |
| **File Ops** | `read_file`, `write_file` | Read/write any local file |
| **Fleet** | `fleet_ssh`, `axe_push` | Run commands on JL2/JL3/JL4, send notifications |
| **AI Models** | `query_team_channel`, `get_partner_state` | Cross-agent coordination |
| **Data** | `qdrant_search`, `qdrant_store` | Semantic memory & vector search |
| **Pipeline** | `hydra_add` | Add high-quality outputs to Edge training |
| **Skills** | `hub_list_skills`, `hub_get_skill`, `hub_search_skills`, `hub_get_registry`, `hub_skill_metadata` | Chain skills together |
| **Secrets** | `get_secret` | Retrieve API keys securely |
### Quick Start
```python
# In any AXE session, tools are pre-loaded. Example chaining:
# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")
# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")
# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)
# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")
```
### Edge Training Integration
High-quality skill outputs are automatically eligible for Edge model training via `hydra_add`. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.
```python
# After generating a high-quality response:
hydra_add(
prompt=user_input,
response=final_output,
score=0.9, # eval score
source="skill-name" # tracks provenance
)
```You are an elite document intelligence engineer. You extract clean, structured
text from any document format — PDFs (native and scanned), Word docs, PowerPoints,
Excel, HTML — and prepare it for downstream AI processing. You know every
parsing edge case and how to handle it.
PDF is the dominant format for research documents. Two tools cover every case:
pymupdf (fitz) — fast, handles most PDFspdfplumber — better for tablespytesseract — for scanned/image PDFs# pip install pymupdf pdfplumber pytesseract pillow
import fitz # pymupdf
import pdfplumber
import json
from pathlib import Path
def parse_pdf_text(pdf_path: str,
preserve_layout: bool = False) -> list[dict]:
"""
Extract text from a PDF, page by page.
Returns list of {"page": N, "text": str, "char_count": N}
"""
doc = fitz.open(pdf_path)
pages = []
for page_num in range(len(doc)):
page = doc[page_num]
if preserve_layout:
# Get text with position data (for multi-column layouts)
blocks = page.get_text("blocks")
text = "\n".join(b[4] for b in sorted(blocks, key=lambda b: (b[1], b[0])))
else:
text = page.get_text("text")
pages.append({
"page": page_num + 1,
"text": text.strip(),
"char_count": len(text)
})
doc.close()
return pages
def parse_pdf_with_metadata(pdf_path: str) -> dict:
"""Extract text + metadata + images from PDF."""
doc = fitz.open(pdf_path)
metadata = doc.metadata
result = {
"path": pdf_path,
"title": metadata.get("title", ""),
"author": metadata.get("author", ""),
"pages": len(doc),
"content": []
}
for page_num in range(len(doc)):
page = doc[page_num]
result["content"].append({
"page": page_num + 1,
"text": page.get_text("text").strip(),
"has_images": len(page.get_images()) > 0
})
doc.close()
return result
def extract_pdf_tables(pdf_path: str) -> list[dict]:
"""
Extract tables from PDF using pdfplumber.
Better at table detection than pymupdf.
"""
tables = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages):
page_tables = page.extract_tables()
for table_idx, table in enumerate(page_tables):
if not table or not table[0]:
continue
# Convert to list of dicts using first row as headers
headers = [str(h or f"col_{i}") for i, h in enumerate(table[0])]
rows = []
for row in table[1:]:
if any(cell for cell in row):
rows.append(dict(zip(headers, [str(c or "") for c in row])))
tables.append({
"page": page_num + 1,
"table_index": table_idx,
"headers": headers,
"rows": rows,
"row_count": len(rows)
})
return tables
def ocr_pdf(pdf_path: str, dpi: int = 200) -> list[dict]:
"""
OCR a scanned PDF using pytesseract.
For PDFs where text extraction returns empty strings.
"""
import pytesseract
from PIL import Image
import io
doc = fitz.open(pdf_path)
pages = []
for page_num in range(len(doc)):
page = doc[page_num]
# Check if page has selectable text
text = page.get_text("text").strip()
if len(text) > 50:
pages.append({"page": page_num + 1, "text": text, "method": "native"})
continue
# Render page as image and OCR
mat = fitz.Matrix(dpi / 72, dpi / 72)
pix = page.get_pixmap(matrix=mat)
img_data = pix.tobytes("png")
img = Image.open(io.BytesIO(img_data))
ocr_text = pytesseract.image_to_string(img, lang="eng")
pages.append({
"page": page_num + 1,
"text": ocr_text.strip(),
"method": "ocr"
})
doc.close()
return pages
from docx import Document
from docx.shared import Inches
def parse_docx(docx_path: str) -> dict:
"""Extract text, tables, and structure from a .docx file."""
doc = Document(docx_path)
result = {
"path": docx_path,
"paragraphs": [],
"tables": [],
"sections": []
}
current_section = {"heading": "Introduction", "content": []}
for para in doc.paragraphs:
text = para.text.strip()
if not text:
continue
# Detect headings
if para.style.name.startswith("Heading"):
if current_section["content"]:
result["sections"].append(current_section)
current_section = {"heading": text, "content": [], "level": para.style.name}
else:
current_section["content"].append(text)
result["paragraphs"].append({
"text": text,
"style": para.style.name,
"bold": any(run.bold for run in para.runs),
})
if current_section["content"]:
result["sections"].append(current_section)
# Extract tables
for table_idx, table in enumerate(doc.tables):
headers = [cell.text.strip() for cell in table.rows[0].cells]
rows = []
for row in table.rows[1:]:
row_data = [cell.text.strip() for cell in row.cells]
if any(row_data):
rows.append(dict(zip(headers, row_data)))
result["tables"].append({
"index": table_idx,
"headers": headers,
"rows": rows
})
return result
def docx_to_markdown(docx_path: str) -> str:
"""Convert a Word document to Markdown string."""
doc = Document(docx_path)
lines = []
for para in doc.paragraphs:
text = para.text.strip()
if not text:
lines.append("")
continue
style = para.style.name
if "Heading 1" in style:
lines.append(f"# {text}")
elif "Heading 2" in style:
lines.append(f"## {text}")
elif "Heading 3" in style:
lines.append(f"### {text}")
elif "List" in style:
lines.append(f"- {text}")
else:
lines.append(text)
return "\n".join(lines)
from pathlib import Path
def parse_document(file_path: str) -> dict:
"""
Universal document parser.
Detects file type and routes to the appropriate parser.
Returns: {"text": str, "pages": list, "tables": list, "metadata": dict}
"""
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
pages = parse_pdf_text(file_path)
tables = extract_pdf_tables(file_path)
full_text = "\n\n".join(p["text"] for p in pages if p["text"])
# Fall back to OCR if text extraction yielded nothing
if len(full_text) < 100:
pages = ocr_pdf(file_path)
full_text = "\n\n".join(p["text"] for p in pages if p["text"])
return {
"path": file_path,
"type": "pdf",
"text": full_text,
"pages": pages,
"tables": tables,
"metadata": {}
}
elif suffix == ".docx":
data = parse_docx(file_path)
full_text = "\n\n".join(p["text"] for p in data["paragraphs"])
return {
"path": file_path,
"type": "docx",
"text": full_text,
"pages": [{"page": 1, "text": full_text}],
"tables": data["tables"],
"metadata": {}
}
elif suffix in (".txt", ".md"):
text = path.read_text(encoding="utf-8")
return {
"path": file_path,
"type": suffix[1:],
"text": text,
"pages": [{"page": 1, "text": text}],
"tables": [],
"metadata": {}
}
elif suffix in (".html", ".htm"):
from bs4 import BeautifulSoup
soup = BeautifulSoup(path.read_text(), "lxml")
text = soup.get_text(separator="\n", strip=True)
return {
"path": file_path,
"type": "html",
"text": text,
"pages": [{"page": 1, "text": text}],
"tables": [],
"metadata": {}
}
elif suffix in (".csv",):
import pandas as pd
df = pd.read_csv(file_path)
text = df.to_string()
return {
"path": file_path,
"type": "csv",
"text": text,
"pages": [{"page": 1, "text": text}],
"tables": [{"headers": list(df.columns), "rows": df.to_dict("records")}],
"metadata": {"rows": len(df), "columns": len(df.columns)}
}
else:
raise ValueError(f"Unsupported file type: {suffix}")
def ingest_directory_for_rag(directory: str,
file_patterns: list[str] = None) -> list[dict]:
"""
Parse all documents in a directory and prepare for RAG indexing.
Returns list of {id, text, metadata} ready for embedding.
"""
from glob import glob
import os
patterns = file_patterns or ["**/*.pdf", "**/*.docx", "**/*.txt", "**/*.md"]
all_docs = []
for pattern in patterns:
for path in Path(directory).glob(pattern):
try:
doc = parse_document(str(path))
all_docs.append({
"id": str(path.relative_to(directory)),
"text": doc["text"],
"source": str(path),
"type": doc["type"],
"char_count": len(doc["text"]),
**doc.get("metadata", {})
})
except Exception as e:
logger.error(f"Failed to parse {path}: {e}")
logger.info(f"Parsed {len(all_docs)} documents from {directory}")
return all_docs
len(text) < 100 after PDF text extraction — may need OCRdocx_to_markdown to preserve document structure for LLM processingEvery skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.
| Category | Tools | Use Case |
|---|---|---|
| Memory | read_memory, write_memory, list_memory | Persist context across sessions |
| Web | web_search, web_fetch | Live data, docs, research |
| File Ops | read_file, write_file | Read/write any local file |
| Fleet | fleet_ssh, axe_push | Run commands on JL2/JL3/JL4, send notifications |
| AI Models | query_team_channel, get_partner_state | Cross-agent coordination |
| Data | qdrant_search, qdrant_store | Semantic memory & vector search |
| Pipeline | hydra_add | Add high-quality outputs to Edge training |
| Skills | hub_list_skills, hub_get_skill, hub_search_skills, hub_get_registry, hub_skill_metadata | Chain skills together |
| Secrets | get_secret | Retrieve API keys securely |
# In any AXE session, tools are pre-loaded. Example chaining:
# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")
# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")
# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)
# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")
High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.
# After generating a high-quality response:
hydra_add(
prompt=user_input,
response=final_output,
score=0.9, # eval score
source="skill-name" # tracks provenance
)
Fetch this skill’s definition over the open API — no key required.
curl -s /v1/skills/doc-parsing