create_prediction_agent_and_get_predictions
Create an AI prediction agent to estimate binary outcome probabilities using expert consensus, confidence intervals, and text/image inputs without historical data.
Instructions
This tool creates a BinaryPredictor agent with your session and input data model and then provides prediction input data to the agent and returns the consensus a prediction from a panel of experts along with their individual estimates and text explanations. The agent also returns the alpha and beta parameters for a Beta distribution that allows you to estimate the confidence interval of its consensus probability estimate.
When to use this tool:
Use this tool to request a probability estimate from Chronulus in situation when there is a binary outcome
This tool is specifically made to estimate the probability of an event occurring and not occurring and does not require historical data
How to use this tool:
First, make sure you have a session_id for the prediction use case.
Next, think about the features / characteristics most suitable for producing the requested prediction and then create an input_data_model that corresponds to the input_data you will provide for the thing or event being predicted.
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, provide the number of experts to consult. The minimum and default number is 2, but users may request up to 30 30 opinions in situations where reproducibility and risk sensitively is of the utmost importance. In most cases, 2 to 5 experts is sufficient.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | The session_id for the forecasting or prediction use case | |
| input_data_model | Yes | Metadata on the fields you will include in the input_data. | |
| 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. | |
| num_experts | Yes | The number of experts to consult when forming consensus |
Implementation Reference
- 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)
- 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'}. """ )
- 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