create_prediction_agent_and_get_predictions

Instructions

Implementation Reference

• src/chronulus_mcp/agent/predictor.py:13-93 (handler)

  The async handler function implementing the tool: loads session, generates and validates input model, creates BinaryPredictor agent, queues and retrieves predictions with expert opinions and beta parameters.

  async def create_prediction_agent_and_get_predictions(
          session_id: Annotated[str, Field(description="The session_id for the forecasting or prediction use case")],
          input_data_model: Annotated[List[InputField], Field(
              description="""Metadata on the fields you will include in the input_data."""
          )],
          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.")],
          ctx: Context,
          num_experts: Annotated[int, Field(description="The number of experts to consult when forming consensus")],
  ) -> Union[str, Dict[str, Union[dict, str]]]:
      """Queues and retrieves a binary event prediction from Chronulus with a predefined session_id
  
      This tool creates a BinaryPredictor agent and then provides a prediction input to the agent and returns the prediction data and
      text explanations from each of the experts consulted by the agent.
  
      Args:
          session_id (str): The session_id for the forecasting or prediction use case.
          input_data_model (List[InputField]): Metadata on the fields you will include in the input_data. Eg., for a field named "brand", add a description like "the brand of the product to forecast"
          input_data (Dict[str, Union[str, dict, List[dict]]]): The prediction 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.
          ctx (Context): Context object providing access to MCP capabilities.
          num_experts (int): The number of experts to consult when forming consensus.
  
      Returns:
          Union[str, Dict[str, Union[dict, str]]]: a dictionary with prediction data, a text explanation of the predictions, agent_id, and probability estimate.
      """
  
  
      try:
          chronulus_session = Session.load_from_saved_session(session_id=session_id, verbose=False)
      except Exception as e:
          error_message = f"Failed to retrieve session with session_id: {session_id}\n\n{e}"
          _ = await ctx.error( message=error_message)
          return error_message
  
      try:
          InputItem = generate_model_from_fields("InputItem", input_data_model)
      except Exception as e:
          error_message = f"Failed to create InputItem model with input data model: {json.dumps(input_data_model, indent=2)}\n\n{e}"
          _ = await ctx.error(message=error_message)
          return error_message
  
      try:
          item = InputItem(**input_data)
      except Exception as e:
          error_message = f"Failed to validate the input_data with the generated InputItem model. \n\n{e}"
          _ = await ctx.error(message=error_message)
          return error_message
  
      try:
          agent = BinaryPredictor(
              session=chronulus_session,
              input_type=InputItem,
              verbose=False,
          )
      except Exception as e:
          return f"""Error at nf_agent: {str(e)}
          
  input_fields = {input_data_model}
  
  input_data = {json.dumps(input_data, indent=2)}
  
  input_type = {str(type(InputItem))}
  """
  
      try:
  
          req = agent.queue(item, num_experts=num_experts, note_length=(5,10))
      except Exception as e:
          return f"""Error at nf_agent: {str(e)}"""
  
      try:
          prediction_set = agent.get_request_predictions(req.request_id)
          return {
              "agent_id": agent.estimator_id,
              "request_id": req.request_id,
              "beta_params": prediction_set.beta_params,
              'expert_opinions': [p.text for p in prediction_set],
              'probability': prediction_set.prob_a}
  
      except Exception as e:
          return f"""Error on prediction: {str(e)}"""

• src/chronulus_mcp/__init__.py:234-234 (registration)

  Registers the create_prediction_agent_and_get_predictions tool with the FastMCP server instance.

  mcp.add_tool(create_prediction_agent_and_get_predictions, description=CREATE_AGENT_AND_GET_PREDICTION_DESCRIPTION)

• src/chronulus_mcp/agent/_types.py:7-18 (schema)

  Pydantic model defining the structure for input_data_model fields, enabling dynamic schema generation for predictions.

  class InputField(BaseModel):
      name: str = Field(description="Field name. Should be a valid python variable name.")
      description: str = Field(description="A description of the value you will pass in the field.")
      type: Literal[
          'str', 'Text', 'List[Text]', 'TextFromFile', 'List[TextFromFile]', 'PdfFromFile', 'List[PdfFromFile]', 'ImageFromFile', 'List[ImageFromFile]'
      ] = Field(
          default='str',
          description="""The type of the field. 
          ImageFromFile takes a single named-argument, 'file_path' as input which should be absolute path to the image to be included. So you should provide this input as json, eg. {'file_path': '/path/to/image'}.
          """
      )

• src/chronulus_mcp/agent/_types.py:26-65 (helper)

  Helper function to dynamically create a Pydantic model from input_data_model for validating input_data in the handler.

  def generate_model_from_fields(model_name: str, fields: List[InputField]) -> Type[BaseModel]:
      """
      Generate a new Pydantic BaseModel from a list of InputField objects.
  
      Args:
          model_name: The name for the generated model class
          fields: List of InputField objects defining the model's fields
  
      Returns:
          A new Pydantic BaseModel class with the specified fields
      """
      literal_type_mapping = {
          'str': str,
          'ImageFromFile': ImageFromFile,
          'List[ImageFromFile]': List[ImageFromFile],
          'TextFromFile': TextFromFile,
          'List[TextFromFile]': List[TextFromFile],
          'PdfFromFile': PdfFromFile,
          'List[PdfFromFile]': List[PdfFromFile]
      }
  
      field_definitions = {
          field.name: (
              Optional[literal_type_mapping.get(field.type, str)],
              Field(description=field.description)
          )
          for field in fields
      }
  
      DynamicModel = create_model(
          model_name,
          __base__=BaseModel,  # Explicitly set BaseModel as the base class
          **field_definitions
      )
  
      DynamicModel.__annotations__ = {
          field.name: str for field in fields
      }
  
      return DynamicModel

• src/chronulus_mcp/__init__.py:7-7 (registration)

  Imports the tool handler from predictor.py for registration.

  from chronulus_mcp.agent.predictor import create_prediction_agent_and_get_predictions, reuse_prediction_agent_and_get_prediction