#!/usr/bin/env python3
"""CLI client for testing the RegisterUZ MCP server."""
import argparse
import asyncio
import json
from fastmcp import Client
from registeruz_mcp import mcp
async def search_company(client, ico=None, dic=None, limit=10):
"""Search for a company by IČO or DIČ."""
params = {"zmenene_od": "2000-01-01", "max_zaznamov": limit}
if ico:
params["ico"] = ico
if dic:
params["dic"] = dic
result = await client.call_tool("get_uctovne_jednotky", params)
return result.data
async def get_company_detail(client, company_id):
"""Get detailed information about a company."""
result = await client.call_tool("get_uctovna_jednotka", {"id": company_id})
return result.data
async def get_accounting_closure(client, closure_id):
"""Get accounting closure details."""
result = await client.call_tool("get_uctovna_zavierka", {"id": closure_id})
return result.data
async def get_financial_report(client, report_id):
"""Get financial report details."""
result = await client.call_tool("get_uctovny_vykaz", {"id": report_id})
return result.data
async def get_template(client, template_id):
"""Get financial report template."""
result = await client.call_tool("get_sablona", {"id": template_id})
return result.data
async def get_labeled_report(client, report_id):
"""Get financial report with labeled tables."""
result = await client.call_tool("get_uctovny_vykaz_with_labeled_tables", {"id": report_id})
return result.data
async def search_report_values(client, report_id, row_label=None, row_code=None, column_label=None, table_name=None):
"""Search for specific values in financial report by labels."""
params = {"id": report_id}
if row_label:
params["row_label"] = row_label
if row_code:
params["row_code"] = row_code
if column_label:
params["column_label"] = column_label
if table_name:
params["table_name"] = table_name
result = await client.call_tool("get_uctovny_vykaz_table_value_by_labels", params)
return result.data
def format_company(company):
"""Format company details for display."""
lines = []
lines.append(f"Company: {getattr(company, 'nazovUJ', 'N/A')}")
lines.append(f" ID: {company.id}")
lines.append(f" IČO: {getattr(company, 'ico', 'N/A')}")
lines.append(f" DIČ: {getattr(company, 'dic', 'N/A')}")
if getattr(company, "ulica", None):
lines.append(
f" Address: {company.ulica}, {getattr(company, 'psc', '')} {getattr(company, 'mesto', '')}"
)
if getattr(company, "datumZalozenia", None):
lines.append(f" Founded: {company.datumZalozenia}")
if getattr(company, "datumZrusenia", None):
lines.append(f" Terminated: {company.datumZrusenia}")
if getattr(company, "pravnaForma", None):
lines.append(f" Legal Form: {company.pravnaForma}")
if getattr(company, "skNace", None):
lines.append(f" SK NACE: {company.skNace}")
closures = getattr(company, "idUctovnychZavierok", None) or []
reports = getattr(company, "idVyrocnychSprav", None) or []
lines.append(f" Accounting Closures: {len(closures)}")
lines.append(f" Annual Reports: {len(reports)}")
return "\n".join(lines)
def format_closure(closure):
"""Format accounting closure for display."""
lines = []
lines.append(f"Accounting Closure ID: {closure.id}")
lines.append(
f" Period: {getattr(closure, 'obdobieOd', 'N/A')} - {getattr(closure, 'obdobieDo', 'N/A')}"
)
lines.append(f" Type: {getattr(closure, 'typ', 'N/A')}")
if getattr(closure, "datumPodania", None):
lines.append(f" Submission Date: {closure.datumPodania}")
reports = getattr(closure, "idUctovnychVykazov", None) or []
lines.append(f" Financial Reports: {len(reports)}")
return "\n".join(lines)
def format_report(report):
"""Format financial report for display."""
lines = []
lines.append(f"Financial Report ID: {report.id}")
lines.append(f" Template ID: {getattr(report, 'idSablony', 'N/A')}")
lines.append(f" Currency: {getattr(report, 'mena', 'N/A')}")
lines.append(f" Accessibility: {getattr(report, 'pristupnostDat', 'N/A')}")
obsah = getattr(report, "obsah", None)
if obsah:
tabulky = getattr(obsah, "tabulky", None) or []
lines.append(f" Tables: {len(tabulky)}")
for i, tabulka in enumerate(tabulky[:5], 1):
nazov = getattr(tabulka, "nazov", None)
if nazov and hasattr(nazov, "sk"):
nazov = nazov.sk
data = getattr(tabulka, "data", None) or []
lines.append(f" [{i}] {nazov} ({len(data)} values)")
return "\n".join(lines)
def format_labeled_report(report):
"""Format financial report with labeled tables for display."""
lines = []
lines.append(f"Financial Report ID: {report.id}")
lines.append(f" Template ID: {getattr(report, 'idSablony', 'N/A')}")
lines.append(f" Currency: {getattr(report, 'mena', 'N/A')}")
tables = getattr(report, "labeled_tables", None) or []
lines.append(f" Labeled Tables: {len(tables)}")
for table in tables:
lines.append(f"\n === {table.table_name} ===")
columns = table.columns
lines.append(f" Columns: {', '.join(columns)}")
for row in table.rows:
code = f"[{row.code}] " if row.code else ""
lines.append(f" {code}{row.label}")
for cv in row.values:
val = cv.value if cv.value else "-"
lines.append(f" {cv.column}: {val}")
return "\n".join(lines)
def format_search_results(results):
"""Format value search results for display."""
lines = []
lines.append(f"Report ID: {results.report_id}")
lines.append(f"Template ID: {getattr(results, 'template_id', 'N/A')}")
matches = getattr(results, "matches", None) or []
lines.append(f"Matches found: {len(matches)}")
for m in matches:
code = f"[{m.row_code}] " if m.row_code else ""
lines.append(f"\n Table: {m.table_name}")
lines.append(f" Row: {code}{m.row_label}")
lines.append(f" Column: {m.column_label}")
lines.append(f" Value: {m.value}")
return "\n".join(lines)
async def main():
parser = argparse.ArgumentParser(
description="Test client for RegisterUZ MCP server (Slovak Registry of Financial Statements)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s --ico 46792511 Search by IČO
%(prog)s --dic 2023579580 Search by DIČ
%(prog)s --ico 46792511 --details Show company details
%(prog)s --ico 46792511 --closures Show accounting closures
%(prog)s --ico 46792511 --latest Show latest financial report
%(prog)s --company-id 1217556 Get company by internal ID
%(prog)s --closure-id 6028050 Get accounting closure by ID
%(prog)s --report-id 4865625 Get financial report by ID
%(prog)s --labeled-report-id 9166748 Get report with labeled tables
%(prog)s --search-report-id 9166748 --row-code A.VI.
Search values by row code
%(prog)s --search-report-id 9166748 --row-label "SPOLU MAJETOK"
Search values by row label
""",
)
# Search options
search_group = parser.add_argument_group("Search Options")
search_group.add_argument("--ico", help="Search by registration number (IČO)")
search_group.add_argument("--dic", help="Search by tax ID (DIČ)")
search_group.add_argument(
"--limit",
type=int,
default=10,
help="Maximum number of search results (default: 10)",
)
# Direct ID lookups
id_group = parser.add_argument_group("Direct ID Lookups")
id_group.add_argument("--company-id", type=int, help="Get company by internal ID")
id_group.add_argument(
"--closure-id", type=int, help="Get accounting closure by ID"
)
id_group.add_argument("--report-id", type=int, help="Get financial report by ID")
id_group.add_argument("--template-id", type=int, help="Get template by ID")
id_group.add_argument("--labeled-report-id", type=int, help="Get financial report with labeled tables")
id_group.add_argument("--search-report-id", type=int, help="Search values in financial report by labels")
# Search value filters
value_group = parser.add_argument_group("Value Search Filters (use with --search-report-id)")
value_group.add_argument("--row-label", help="Partial row label to match (case-insensitive)")
value_group.add_argument("--row-code", help="Row designation code to match (e.g., 'A.', 'A.I.', 'A.VI.')")
value_group.add_argument("--column-label", help="Partial column label to match (case-insensitive)")
value_group.add_argument("--table-name", help="Partial table name to filter (case-insensitive)")
# Output options
output_group = parser.add_argument_group("Output Options")
output_group.add_argument(
"--details",
action="store_true",
help="Show detailed company information",
)
output_group.add_argument(
"--closures",
action="store_true",
help="Show accounting closures for the company",
)
output_group.add_argument(
"--latest",
action="store_true",
help="Show latest financial report",
)
output_group.add_argument(
"--json",
action="store_true",
dest="output_json",
help="Output raw JSON response",
)
args = parser.parse_args()
# Validate arguments
if not any(
[
args.ico,
args.dic,
args.company_id,
args.closure_id,
args.report_id,
args.template_id,
args.labeled_report_id,
args.search_report_id,
]
):
parser.error("Please provide at least one of: --ico, --dic, --company-id, --closure-id, --report-id, --template-id, --labeled-report-id, --search-report-id")
async with Client(mcp) as client:
# Handle labeled report first
if args.labeled_report_id:
print(f"[Tool: get_uctovny_vykaz_with_labeled_tables(id={args.labeled_report_id})]\n")
report = await get_labeled_report(client, args.labeled_report_id)
if args.output_json:
if hasattr(report, "model_dump"):
data = report.model_dump()
elif hasattr(report, "__dict__"):
data = {k: v for k, v in vars(report).items() if not k.startswith("_")}
else:
data = report
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
else:
print(format_labeled_report(report))
return
# Handle search report values
if args.search_report_id:
params = f"id={args.search_report_id}"
if args.row_label:
params += f", row_label='{args.row_label}'"
if args.row_code:
params += f", row_code='{args.row_code}'"
if args.column_label:
params += f", column_label='{args.column_label}'"
if args.table_name:
params += f", table_name='{args.table_name}'"
print(f"[Tool: get_uctovny_vykaz_table_value_by_labels({params})]\n")
results = await search_report_values(
client,
args.search_report_id,
row_label=args.row_label,
row_code=args.row_code,
column_label=args.column_label,
table_name=args.table_name,
)
if args.output_json:
if hasattr(results, "model_dump"):
data = results.model_dump()
elif hasattr(results, "__dict__"):
data = {k: v for k, v in vars(results).items() if not k.startswith("_")}
else:
data = results
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
else:
print(format_search_results(results))
return
# Handle direct ID lookups first
if args.closure_id:
print(f"[Tool: get_uctovna_zavierka(id={args.closure_id})]\n")
closure = await get_accounting_closure(client, args.closure_id)
if args.output_json:
if hasattr(closure, "model_dump"):
data = closure.model_dump()
elif hasattr(closure, "__dict__"):
data = {k: v for k, v in vars(closure).items() if not k.startswith("_")}
else:
data = closure
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
else:
print(format_closure(closure))
return
if args.report_id:
print(f"[Tool: get_uctovny_vykaz(id={args.report_id})]\n")
report = await get_financial_report(client, args.report_id)
if args.output_json:
if hasattr(report, "model_dump"):
data = report.model_dump()
elif hasattr(report, "__dict__"):
data = {k: v for k, v in vars(report).items() if not k.startswith("_")}
else:
data = report
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
else:
print(format_report(report))
return
if args.template_id:
print(f"[Tool: get_sablona(id={args.template_id})]\n")
template = await get_template(client, args.template_id)
if args.output_json:
if hasattr(template, "model_dump"):
data = template.model_dump()
elif hasattr(template, "__dict__"):
data = {k: v for k, v in vars(template).items() if not k.startswith("_")}
else:
data = template
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
else:
print(f"Template: {getattr(template, 'nazov', 'N/A')}")
print(f" ID: {template.id}")
print(f" Regulation: {getattr(template, 'nariadenieMF', 'N/A')}")
print(f" Valid From: {getattr(template, 'platneOd', 'N/A')}")
print(f" Valid To: {getattr(template, 'platneDo', 'N/A')}")
tabulky = getattr(template, "tabulky", None) or []
print(f" Tables: {len(tabulky)}")
return
# Get company ID (either directly or by search)
company_id = args.company_id
if not company_id:
search_params = f"zmenene_od='2000-01-01', max_zaznamov={args.limit}"
if args.ico:
search_params += f", ico='{args.ico}'"
if args.dic:
search_params += f", dic='{args.dic}'"
print(f"[Tool: get_uctovne_jednotky({search_params})]\n")
search_result = await search_company(
client, ico=args.ico, dic=args.dic, limit=args.limit
)
ids = getattr(search_result, "id", [])
if not ids:
print("No companies found matching the search criteria.")
return
if len(ids) > 1 and not args.details:
print(f"Found {len(ids)} companies. IDs: {ids}")
print("Use --company-id to select a specific company, or --details to show all.")
if args.output_json:
print(json.dumps({"ids": ids}, indent=2))
return
company_id = ids[0]
print(f"[Tool: get_uctovna_jednotka(id={company_id})]\n")
else:
print(f"[Tool: get_uctovna_jednotka(id={company_id})]\n")
# Get company details
company = await get_company_detail(client, company_id)
if args.output_json and not (args.closures or args.latest):
# Convert Root object to dict
if hasattr(company, "model_dump"):
data = company.model_dump()
elif hasattr(company, "__dict__"):
data = {k: v for k, v in vars(company).items() if not k.startswith("_")}
else:
data = company
print(
json.dumps(
data,
indent=2,
ensure_ascii=False,
default=str,
)
)
return
print(format_company(company))
# Show closures if requested
if args.closures or args.latest:
closure_ids = getattr(company, "idUctovnychZavierok", None) or []
if not closure_ids:
print("\nNo accounting closures found for this company.")
return
if args.latest:
# Get most recent closures and find the one with latest period
print(f"\n[Tool: get_uctovna_zavierka() x {min(5, len(closure_ids))} calls]")
print("--- Most Recent Accounting Closures ---")
closures = []
for cid in closure_ids[-5:]: # Check last 5
closure = await get_accounting_closure(client, cid)
closures.append(closure)
# Sort by period end date (if available)
closures.sort(
key=lambda c: getattr(c, "obdobieDo", "") or "", reverse=True
)
latest = closures[0] if closures else None
if latest:
print(format_closure(latest))
# Get financial report
report_ids = getattr(latest, "idUctovnychVykazov", None) or []
if report_ids:
print(f"\n[Tool: get_uctovny_vykaz(id={report_ids[0]})]")
print("--- Latest Financial Report ---")
report = await get_financial_report(client, report_ids[0])
print(format_report(report))
else:
print(f"\n[Tool: get_uctovna_zavierka() x {len(closure_ids)} calls]")
print("--- All Accounting Closures ---")
for cid in closure_ids:
closure = await get_accounting_closure(client, cid)
print(format_closure(closure))
print()
if __name__ == "__main__":
asyncio.run(main())
