Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 113 additions & 16 deletions filozzy-mcp/filozzy_mcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,13 +85,71 @@ def _build_session() -> requests.Session:
return session


def _build_display_items(items: list[dict]) -> list[dict]:
"""Strip internal fields and empty values from items for output."""
return [
{k: v for k, v in item.items() if not k.startswith("_") and v not in (None, "")}
for item in items
]


def _format_json(
display_items: list[dict],
has_more: bool,
next_cursor: Optional[str],
) -> str:
"""Return a JSON object with items array and pagination metadata."""
payload: dict = {"items": display_items, "total_in_page": len(display_items)}
if has_more:
payload["has_more"] = True
payload["next_cursor"] = next_cursor
return json.dumps(payload, ensure_ascii=False)


def _format_compact(
display_items: list[dict],
has_more: bool,
next_cursor: Optional[str],
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think compact pagination can still produce incompatible schemas across pages here. Since _build_display_items() strips empty values before _format_compact() builds columns, page 1 and page 2 can end up with different column sets for the same requested fields.

The playbook merge example keeps page 1’s .columns and just concatenates rows, so if a field is empty on page 1 but present on page 2, reconstruction can silently drop or mislabel values.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed — I have a follow-up PR that discards compact entirely, so I won't fix this now.

) -> str:
"""Return columnar JSON: column names once, then rows as arrays.

Much more token-efficient than ``_format_json`` for large result sets
because field names appear once instead of once-per-item. The output
is still valid JSON and can be converted back to objects with jq::

jq '[.columns as $c | .rows[] | [$c, .] | transpose | map({(.[0]): .[1]}) | add]'
"""
# Build a stable column order from the union of all items' keys,
# preserving first-seen order (which follows the requested fields).
columns: list[str] = []
seen: set[str] = set()
for item in display_items:
for key in item:
if key not in seen:
columns.append(key)
seen.add(key)

rows = [[item.get(col, "") for col in columns] for item in display_items]

payload: dict = {
"columns": columns,
"rows": rows,
"total_in_page": len(display_items),
}
if has_more:
payload["has_more"] = True
payload["next_cursor"] = next_cursor
return json.dumps(payload, ensure_ascii=False)


@mcp.tool()
def list_board_items(
query: str = '-status:"🎉 Done"',
fields: Optional[str] = None,
per_page: int = 50,
cursor: Optional[str] = None,
verbose: bool = False,
format: Optional[str] = None,
) -> str:
"""List project board items, filtered via the `query` parameter.

Expand All @@ -118,12 +176,34 @@ def list_board_items(
Default: Repository, Id, url, Title, Status, Kind,
Milestone, Assignees, Cycle Theme, Dev Days Estimate.
Use list_board_fields to see available fields.
In addition to board fields, the following pseudo-fields
are available (derived from item metadata, not board columns):
Node ID — project item node ID (e.g., "PVTI_..."), needed
for set_board_item_field / bulk_set_board_item_field.
Repository, Id, Number, url, Title, Kind, Assignees —
built-in item properties (some included by default).
per_page: Number of items per page (default: 50, max: 100).
cursor: Opaque cursor from a previous response to fetch the next page.
When provided, the same query and fields from the original
request are used automatically.
verbose: If true, include debug info showing the raw REST query,
endpoint, requested field IDs, and item counts.
format: Output format. Default (None) returns human-readable JSONL
with a "Found N items:" header line — designed for LLM
readability in conversation.
"json" returns a single JSON object:
{"items": [...], "total_in_page": N}
{"items": [...], "total_in_page": N, "has_more": true, "next_cursor": "..."}
"compact" returns columnar JSON — field names once, rows as arrays:
{"columns": ["Repository", "Title", ...],
"rows": [["curio", "Fix X", ...], ...],
"total_in_page": N}
Much more token-efficient than "json" for large result sets
(~40-60% smaller) because field names appear once instead of
once-per-item. Convert back to objects with jq:
jq '[.columns as $c | .rows[] | [$c, .] | transpose | map({(.[0]): .[1]}) | add]'
**Recommended for sweep automation** — use this when writing
results to disk for programmatic processing.

Query syntax reference (passed as the REST API `q` parameter):

Expand Down Expand Up @@ -271,20 +351,26 @@ def list_board_items(
next_cursor = result["next_cursor"]
has_more = result["has_more"]

_VALID_FORMATS = {None, "json", "compact"}
if format not in _VALID_FORMATS:
raise ValueError(
f'Unknown format {format!r}. Must be one of: "json", "compact", or omitted for default JSONL.'
)

display_items = _build_display_items(items) if items else []

if format == "json":
return _format_json(display_items, has_more, next_cursor)
if format == "compact":
return _format_compact(display_items, has_more, next_cursor)

if not items:
msg = f"No items found matching query: {query}"
if verbose:
msg += f"\n\n--- Debug ---\n{json.dumps(debug, indent=2)}"
Comment thread
BigLep marked this conversation as resolved.
return msg

lines = []
for item in items:
display = {
k: v
for k, v in item.items()
if not k.startswith("_") and v not in (None, "")
}
lines.append(json.dumps(display, ensure_ascii=False))
lines = [json.dumps(d, ensure_ascii=False) for d in display_items]

header = f"Found {len(items)} items"
if has_more:
Expand All @@ -309,6 +395,7 @@ def list_board_view_items(
per_page: int = 50,
cursor: Optional[str] = None,
verbose: bool = False,
format: Optional[str] = None,
) -> str:
"""List project items for a GitHub project view URL.

Expand All @@ -329,7 +416,17 @@ def list_board_view_items(
per_page: Number of items per page (default: 50, max: 100).
cursor: Opaque cursor from a previous response to fetch the next page.
verbose: If true, include resolved view/filter debug details.
format: Output format. Default (None) returns human-readable JSONL.
"json" returns a single JSON object with an "items" array
and pagination metadata (see list_board_items for details).
"compact" returns columnar JSON (see list_board_items).
"""
_VALID_FORMATS = {None, "json", "compact"}
if format not in _VALID_FORMATS:
raise ValueError(
f'Unknown format {format!r}. Must be one of: "json", "compact", or omitted for default JSONL.'
)

session = _build_session()
resolved = resolve_view_url(session, view_url=view_url)

Expand Down Expand Up @@ -360,6 +457,13 @@ def list_board_view_items(
next_cursor = result["next_cursor"]
has_more = result["has_more"]

display_items = _build_display_items(items) if items else []

Comment thread
BigLep marked this conversation as resolved.
if format == "json":
return _format_json(display_items, has_more, next_cursor)
if format == "compact":
return _format_compact(display_items, has_more, next_cursor)

if not items:
msg = f"No items found for view URL: {view_url}"
if order_warning:
Expand All @@ -373,14 +477,7 @@ def list_board_view_items(
)
return msg

lines = []
for item in items:
display = {
k: v
for k, v in item.items()
if not k.startswith("_") and v not in (None, "")
}
lines.append(json.dumps(display, ensure_ascii=False))
lines = [json.dumps(d, ensure_ascii=False) for d in display_items]

header = f"Found {len(items)} items for view #{resolved['view_number']} ({resolved['view_name']})"
if has_more:
Expand Down
166 changes: 166 additions & 0 deletions filozzy-mcp/tests/test_format.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
"""Unit tests for output format helpers (_format_json, _format_compact).

These are pure-logic tests — no network access or GITHUB_TOKEN required.

Run:
cd filozzy-mcp
uv run pytest tests/test_format.py -v
"""

from __future__ import annotations

import json

from filozzy_mcp.server import _format_json, _format_compact


# ---------------------------------------------------------------------------
# Fixtures / helpers
# ---------------------------------------------------------------------------

SAMPLE_ITEMS = [
{
"Repository": "curio",
"Title": "Fix X",
"Status": "⌨️ In Progress",
"Node ID": "PVTI_abc",
"Assignees": "alice",
},
{
"Repository": "dealbot",
"Title": "Add Y",
"Status": "📌 Triage",
# Node ID and Assignees intentionally missing (sparse row)
},
{
"Repository": "curio",
"Title": "Refactor Z",
"Status": "👀 Awaiting review",
"Node ID": "PVTI_xyz",
"Assignees": "bob, carol",
},
]


def _parse(result: str) -> dict:
return json.loads(result)


# ---------------------------------------------------------------------------
# _format_json
# ---------------------------------------------------------------------------


class TestFormatJson:
def test_basic_structure(self):
out = _parse(_format_json(SAMPLE_ITEMS, has_more=False, next_cursor=None))
assert out["total_in_page"] == 3
assert len(out["items"]) == 3
assert "has_more" not in out
assert "next_cursor" not in out

def test_items_are_original_dicts(self):
out = _parse(_format_json(SAMPLE_ITEMS, has_more=False, next_cursor=None))
assert out["items"][0]["Repository"] == "curio"
assert out["items"][1]["Title"] == "Add Y"

def test_pagination_fields(self):
out = _parse(_format_json(SAMPLE_ITEMS, has_more=True, next_cursor="cur_42"))
assert out["has_more"] is True
assert out["next_cursor"] == "cur_42"

def test_empty_items(self):
out = _parse(_format_json([], has_more=False, next_cursor=None))
assert out["items"] == []
assert out["total_in_page"] == 0


# ---------------------------------------------------------------------------
# _format_compact
# ---------------------------------------------------------------------------


class TestFormatCompact:
def test_columns_are_union_of_all_keys(self):
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
assert set(out["columns"]) == {
"Repository",
"Title",
"Status",
"Node ID",
"Assignees",
}

def test_column_order_follows_first_seen(self):
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
# First item's keys define the initial order
cols = out["columns"]
assert cols.index("Repository") < cols.index("Title")
assert cols.index("Title") < cols.index("Status")

def test_row_count_matches_items(self):
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
assert len(out["rows"]) == 3
assert out["total_in_page"] == 3

def test_row_values_match_items(self):
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
cols = out["columns"]
first_row = dict(zip(cols, out["rows"][0]))
assert first_row["Repository"] == "curio"
assert first_row["Title"] == "Fix X"
assert first_row["Node ID"] == "PVTI_abc"

def test_sparse_rows_get_empty_string(self):
"""Items missing a column that other items have should get ''."""
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
cols = out["columns"]
second_row = dict(zip(cols, out["rows"][1]))
assert second_row["Node ID"] == ""
assert second_row["Assignees"] == ""

def test_pagination_fields(self):
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=True, next_cursor="cur_99"))
assert out["has_more"] is True
assert out["next_cursor"] == "cur_99"

def test_no_pagination_fields_when_not_needed(self):
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
assert "has_more" not in out
assert "next_cursor" not in out

def test_empty_items(self):
out = _parse(_format_compact([], has_more=False, next_cursor=None))
assert out["columns"] == []
assert out["rows"] == []
assert out["total_in_page"] == 0

def test_roundtrip_via_jq_reconstruction(self):
"""Verify the documented jq reconstruction produces the original items."""
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
cols = out["columns"]
# Simulate: jq '[.columns as $c | .rows[] | [$c, .] | transpose | map({(.[0]): .[1]}) | add]'
reconstructed = [
{col: val for col, val in zip(cols, row)} for row in out["rows"]
]
# First and third items should match exactly
assert reconstructed[0] == SAMPLE_ITEMS[0]
assert reconstructed[2] == SAMPLE_ITEMS[2]
# Second item had missing fields — reconstruction has "" instead
for key in SAMPLE_ITEMS[1]:
assert reconstructed[1][key] == SAMPLE_ITEMS[1][key]

def test_compact_is_smaller_than_json(self):
"""The whole point: compact should use fewer bytes than json."""
json_out = _format_json(SAMPLE_ITEMS, has_more=False, next_cursor=None)
compact_out = _format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None)
assert len(compact_out) < len(json_out), (
f"compact ({len(compact_out)}) should be smaller than json ({len(json_out)})"
)

def test_unicode_preserved(self):
"""Emoji status values should survive the round-trip."""
out = _parse(_format_compact(SAMPLE_ITEMS, has_more=False, next_cursor=None))
cols = out["columns"]
first_row = dict(zip(cols, out["rows"][0]))
assert first_row["Status"] == "⌨️ In Progress"
Loading
Loading