reuse_forecasting_agent_and_get_forecast
Generate forecasts for values between 0 and 1 using a NormalizedForecaster agent. Input session data, provide relevant features, and specify forecast horizon and time scale to receive predictions and explanations. Ideal for seasonal weights, probabilities, or share forecasts without historical data.
Instructions
This tool creates a NormalizedForecaster agent with your session and input data model and then provides a forecast input data to the agent and returns the prediction data and text explanation from the agent.
When to use this tool:
Use this tool to request a forecast from Chronulus
This tool is specifically made to forecast values between 0 and 1 and does not require historical data
The prediction can be thought of as seasonal weights, probabilities, or shares of something as in the decimal representation of a percent
How to use this tool:
First, make sure you have a session_id for the forecasting or prediction use case.
Next, think about the features / characteristics most suitable for producing the requested forecast and then create an input_data_model that corresponds to the input_data you will provide for the thing being forecasted.
Remember to pass all relevant information to Chronulus including text and images provided by the user.
If a user gives you files about a thing you are forecasting or predicting, you should pass these as inputs to the agent using one of the following types:
ImageFromFile
List[ImageFromFile]
TextFromFile
List[TextFromFile]
PdfFromFile
List[PdfFromFile]
If you have a large amount of text (over 500 words) to pass to the agent, you should use the Text or List[Text] field types
Finally, add information about the forecasting horizon and time scale requested by the user
Assume the dates and datetimes in the prediction results are already converted to the appropriate local timezone if location is a factor in the use case. So do not try to convert from UTC to local time when plotting.
When plotting the predictions, use a Rechart time series with the appropriate axes labeled and with the prediction explanation displayed as a caption below the plot
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| agent_id | Yes | The agent_id for the forecasting or prediction use case and previously defined input_data_model | |
| forecast_start_dt_str | Yes | The datetime str in '%Y-%m-%d %H:%M:%S' format of the first value in the forecast horizon. | |
| horizon_len | No | The integer length of the forecast horizon. Eg., 60 if a 60 day forecast was requested. | |
| input_data | Yes | The forecast inputs that you will pass to the chronulus agent to make the prediction. The keys of the dict should correspond to the InputField name you provided in input_fields. | |
| time_scale | No | The times scale of the forecast horizon. Valid time scales are 'hours', 'days', and 'weeks'. | days |
Implementation Reference
- The handler function that executes the tool: loads a saved NormalizedForecaster agent by agent_id, instantiates input item, queues a forecast over the specified horizon, retrieves the prediction, and returns agent_id, prediction_id, forecast data as JSON, and textual explanation. Includes input schema via Pydantic Annotated Fields.async def reuse_forecasting_agent_and_get_forecast( agent_id: Annotated[str, Field(description="The agent_id for the forecasting or prediction use case and previously defined input_data_model")], input_data: Annotated[Dict[str, Union[str, dict, List[dict]]], Field( description="The forecast inputs that you will pass to the chronulus agent to make the prediction. The keys of the dict should correspond to the InputField name you provided in input_fields.")], forecast_start_dt_str: Annotated[str, Field( description="The datetime str in '%Y-%m-%d %H:%M:%S' format of the first value in the forecast horizon.")], time_scale: Annotated[str, Field( description="The times scale of the forecast horizon. Valid time scales are 'hours', 'days', and 'weeks'.", default="days")], horizon_len: Annotated[int, Field( description="The integer length of the forecast horizon. Eg., 60 if a 60 day forecast was requested.", default=60)], ) -> Union[str, Dict[str, Union[dict, str]]]: """Queues and retrieves a forecast from Chronulus with a previously created agent_id This tool provides a forecast input to a previous created Chronulus NormalizedForecaster agent and returns the prediction data and text explanation from the agent. Args: agent_id (str): The agent_id for the forecasting or prediction use case and previously defined input_data_model input_data (Dict[str, Union[str, dict, List[dict]]]): The forecast inputs that you will pass to the chronulus agent to make the prediction. The keys of the dict should correspond to the InputField name you provided in input_fields. forecast_start_dt_str (str): The datetime str in '%Y-%m-%d %H:%M:%S' format of the first value in the forecast horizon." time_scale (str): The times scale of the forecast horizon. Valid time scales are 'hours', 'days', and 'weeks'. horizon_len (int): The integer length of the forecast horizon. Eg., 60 if a 60 day forecast was requested. Returns: Union[str, Dict[str, Union[dict, str]]]: a dictionary with prediction data, a text explanation of the predictions, agent_id, and the prediction id. """ nf_agent = NormalizedForecaster.load_from_saved_estimator(estimator_id=agent_id, verbose=False) item = nf_agent.input_type(**input_data) try: forecast_start_dt = datetime.fromisoformat(forecast_start_dt_str) horizon_params = { 'start_dt': forecast_start_dt, time_scale: horizon_len } req = nf_agent.queue(item, **horizon_params) except Exception as e: return f"""Error at nf_agent: {str(e)}""" try: predictions = nf_agent.get_predictions(req.request_id) prediction = predictions[0] return { "agent_id": nf_agent.estimator_id, "prediction_id": prediction.id, 'data': prediction.to_json(orient='rows'), 'explanation': prediction.text} except Exception as e: return f"""Error on prediction: {str(e)}"""
- src/chronulus_mcp/__init__.py:154-154 (registration)Registers the tool with the FastMCP server instance, providing a description (note: uses CREATE_AGENT description).mcp.add_tool(reuse_forecasting_agent_and_get_forecast, description=CREATE_AGENT_AND_GET_FORECAST_DESCRIPTION)
- src/chronulus_mcp/__init__.py:6-6 (registration)Imports the handler function into the main module for registration.from chronulus_mcp.agent.forecaster import create_forecasting_agent_and_get_forecast, reuse_forecasting_agent_and_get_forecast, rescale_forecast