feat(tools): add get_datasets, get_reports, and export_report functions to Power BI integration

This commit is contained in:
Your hh3538962
2026-02-11 22:19:51 +05:00
parent b455c8a2ad
commit 765f7cae58
2 changed files with 170 additions and 8 deletions
@@ -1,13 +1,77 @@
# Power BI Tool
Power BI integration for dataset refresh and report management.
Power BI integration for automated dataset refresh, report export, and workspace management.
## Features
- Get list of workspaces
- Refresh datasets
- Export reports (coming soon)
- **Workspace Management**: List all available Power BI workspaces
- **Dataset Operations**: List and refresh datasets programmatically
- **Report Management**: List reports and export to PDF/PowerPoint
- **OAuth2 Authentication**: Secure credential storage via Hive credential system
## Available Functions
### `power_bi_get_workspaces()`
Get list of all Power BI workspaces accessible to the authenticated user.
### `power_bi_get_datasets(workspace_id)`
List all datasets in a specific workspace.
**Args:**
- `workspace_id`: The Power BI workspace ID
### `power_bi_get_reports(workspace_id)`
List all reports in a specific workspace.
**Args:**
- `workspace_id`: The Power BI workspace ID
### `power_bi_refresh_dataset(workspace_id, dataset_id)`
Trigger a dataset refresh.
**Args:**
- `workspace_id`: The Power BI workspace ID
- `dataset_id`: The dataset ID to refresh
### `power_bi_export_report(workspace_id, report_id, format)`
Export a report to PDF or PowerPoint.
**Args:**
- `workspace_id`: The Power BI workspace ID
- `report_id`: The report ID to export
- `format`: Export format - "PDF" or "PPTX" (default: "PDF")
## Authentication
Requires Power BI access token via credential store or environment variable.
Requires Power BI access token obtained via OAuth2.
**Via Credential Store:**
```python
# Token stored securely in Hive credential store under key "power_bi"
```
**Via Environment Variable:**
```bash
export POWER_BI_ACCESS_TOKEN="your-token-here"
```
## Example Use Cases
**Business Analytics Agent:**
```
1. Monthly sales analysis completes
2. Agent calls power_bi_refresh_dataset() to update dashboard
3. Agent calls power_bi_export_report() to generate PDF
4. Report emailed to stakeholders automatically
```
**Marketing Agent:**
```
1. Identifies trending products from data analysis
2. Calls power_bi_refresh_dataset() on marketing dashboard
3. Team receives updated insights in real-time
```
## API Reference
Based on Power BI REST API v1.0: https://learn.microsoft.com/en-us/rest/api/power-bi/
@@ -72,6 +72,39 @@ class _PowerBIClient:
)
return self._handle_response(response)
def get_datasets(self, workspace_id: str) -> dict[str, Any]:
"""Get list of datasets in a workspace."""
response = httpx.get(
f"{POWER_BI_API_BASE}/groups/{workspace_id}/datasets",
headers=self._headers,
timeout=30.0,
)
return self._handle_response(response)
def get_reports(self, workspace_id: str) -> dict[str, Any]:
"""Get list of reports in a workspace."""
response = httpx.get(
f"{POWER_BI_API_BASE}/groups/{workspace_id}/reports",
headers=self._headers,
timeout=30.0,
)
return self._handle_response(response)
def export_report(
self, workspace_id: str, report_id: str, format: str = "PDF"
) -> dict[str, Any]:
"""Export a report to PDF or PPTX."""
if format.upper() not in ["PDF", "PPTX"]:
return {"error": f"Invalid format: {format}. Must be PDF or PPTX"}
response = httpx.post(
f"{POWER_BI_API_BASE}/groups/{workspace_id}/reports/{report_id}/Export",
headers=self._headers,
json={"format": format.upper()},
timeout=30.0,
)
return self._handle_response(response)
def register_tools(
mcp: FastMCP,
@@ -98,8 +131,6 @@ def register_tools(
}
return _PowerBIClient(token)
# --- Workspaces ---
@mcp.tool()
def power_bi_get_workspaces() -> dict:
"""
@@ -138,4 +169,71 @@ def register_tools(
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": "Network error: {e}"}
return {"error": f"Network error: {e}"}
@mcp.tool()
def power_bi_get_datasets(workspace_id: str) -> dict:
"""
Get list of datasets in a Power BI workspace.
Args:
workspace_id: The Power BI workspace ID
Returns:
Dict with list of datasets or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.get_datasets(workspace_id)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": f"Network error: {e}"}
@mcp.tool()
def power_bi_get_reports(workspace_id: str) -> dict:
"""
Get list of reports in a Power BI workspace.
Args:
workspace_id: The Power BI workspace ID
Returns:
Dict with list of reports or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.get_reports(workspace_id)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": f"Network error: {e}"}
@mcp.tool()
def power_bi_export_report(
workspace_id: str, report_id: str, format: str = "PDF"
) -> dict:
"""
Export a Power BI report to PDF or PPTX.
Args:
workspace_id: The Power BI workspace ID
report_id: The report ID to export
format: Export format - "PDF" or "PPTX" (default: "PDF")
Returns:
Dict with export result or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.export_report(workspace_id, report_id, format)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": f"Network error: {e}"}