write_data
Enables writing timeseries data to a specified database using InfluxDB Line Protocol format, supporting seamless integration with GigAPI Timeseries Lake for efficient data management.
Instructions
Write data using InfluxDB Line Protocol format.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| data | Yes | ||
| database | Yes |
Implementation Reference
- mcp_gigapi/tools.py:248-252 (registration)Registration of the 'write_data' MCP tool using FastMCP's Tool.from_function, binding to GigAPITools.write_data method.Tool.from_function( tools_instance.write_data, name="write_data", description="Write data using InfluxDB Line Protocol format.", ),
- mcp_gigapi/tools.py:153-177 (handler)Handler function for the 'write_data' tool in GigAPITools class, which delegates to the client and wraps the response.def write_data(self, database: str, data: str) -> Dict[str, Any]: """Write data using InfluxDB Line Protocol. Args: database: The database to write to data: Data in InfluxDB Line Protocol format Returns: Write operation result """ try: result = self.client.write_data(database, data) return { "result": result, "success": True, "database": database, "data_lines": len(data.strip().split('\n')) } except GigAPIClientError as e: logger.error(f"Failed to write data: {e}") return { "error": str(e), "success": False, "database": database }
- mcp_gigapi/tools.py:27-32 (schema)Pydantic schema/model for 'write_data' tool input validation (database and data fields).class WriteDataInput(BaseModel): """Input model for data writing operations.""" database: str = Field(..., description="The database to write to") data: str = Field(..., description="Data in InfluxDB Line Protocol format")
- mcp_gigapi/client.py:167-188 (helper)Core implementation in GigAPIClient that performs the HTTP POST to /write endpoint for data ingestion.def write_data(self, database: str, data: str) -> Dict[str, Any]: """Write data using InfluxDB Line Protocol. Args: database: Database name data: Data in InfluxDB Line Protocol format Returns: Write response """ params = {"db": database} headers = {"Content-Type": "text/plain"} response = self._make_request( "POST", "/write", data=data, params=params, headers=headers ) return response.json()