Chronulus MCP Server

Instructions

Implementation Reference

• src/chronulus_mcp/agent/forecaster.py:106-158 (handler)

  The handler function for the 'reuse_forecasting_agent_and_get_forecast' tool. It loads a pre-existing NormalizedForecaster agent using the provided agent_id, validates and processes the input_data, queues a forecast over the specified horizon, retrieves the prediction, and returns a dictionary containing the agent_id, prediction_id, forecast data, and textual explanation.

  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 'reuse_forecasting_agent_and_get_forecast' tool with the FastMCP server instance. Note that it uses the description constant intended for the create tool.

  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 'reuse_forecasting_agent_and_get_forecast' handler from the forecaster module to make it available for registration.

  from chronulus_mcp.agent.forecaster import create_forecasting_agent_and_get_forecast, reuse_forecasting_agent_and_get_forecast, rescale_forecast

• src/chronulus_mcp/__init__.py:94-111 (schema)

  Defines the description string for the reuse forecasting tool, detailing usage instructions and parameters, though not directly used in the registration (uses create tool's description instead). Serves as tool schema/documentation.

  REUSE_AGENT_AND_GET_FORECAST_DESCRIPTION = f"""
  This tool provides a forecast input to a previous created Chronulus NormalizedForecaster 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 a Chronulus agent that you have already created and when your input data model is unchanged
  - 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 an agent_id for the agent. The agent is already attached to the correct session. So you do not need to provide a session_id.
  - Next, reference the input data model that you previously used with the agent and create new inputs for the item being forecast
      that align with the previously specified input data model
  {FILE_TYPE_INSTRUCTIONS}
  - 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
  """