count_rows
Count rows in PostgreSQL tables for Avanti Fellows database, with optional filtering using WHERE clauses to analyze data volume.
Instructions
Count rows in a table, optionally with a WHERE clause.
Args:
table_name: Name of the table
schema_name: Schema name (default: public)
where: Optional WHERE clause (without 'WHERE' keyword)
Returns:
JSON with countInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| table_name | Yes | ||
| schema_name | No | public | |
| where | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- mcp_postgres/server.py:205-230 (handler)The handler function for the 'count_rows' MCP tool. It is registered via the @mcp.tool() decorator. Constructs and executes a safe COUNT(*) query on the specified table, optionally with a validated WHERE clause, and returns the row count as JSON.
@mcp.tool() async def count_rows(table_name: str, schema_name: str = "public", where: str = None) -> str: """Count rows in a table, optionally with a WHERE clause. Args: table_name: Name of the table schema_name: Schema name (default: public) where: Optional WHERE clause (without 'WHERE' keyword) Returns: JSON with count """ sql = f'SELECT COUNT(*) as count FROM "{schema_name}"."{table_name}"' if where: # Basic validation - only allow read operations in WHERE if not is_read_only(f"SELECT * FROM t WHERE {where}"): return json.dumps({"error": "Invalid WHERE clause"}) sql += f" WHERE {where}" try: async with get_connection() as conn: row = await conn.fetchrow(sql) return json.dumps({"count": row["count"]}) except Exception as e: return json.dumps({"error": str(e)})