create_prediction_agent_and_get_predictions
Create a BinaryPredictor agent to estimate the probability of a binary event using expert consensus, individual estimates, and confidence intervals. Provide session ID, input data model, and expert count for predictions 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 |
|---|---|---|---|
| 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. | |
| input_data_model | Yes | Metadata on the fields you will include in the input_data. | |
| num_experts | Yes | The number of experts to consult when forming consensus | |
| session_id | Yes | The session_id for the forecasting or prediction use case |
Implementation Reference
- The main asynchronous handler function that implements the tool logic. It loads a Chronulus session, generates and validates an input model from input_data_model, creates a BinaryPredictor agent, queues a prediction request with the specified number of experts, retrieves the prediction set, and returns a dictionary containing agent_id, request_id, beta_params, expert_opinions, and probability.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 using mcp.add_tool, providing a detailed description.mcp.add_tool(create_prediction_agent_and_get_predictions, description=CREATE_AGENT_AND_GET_PREDICTION_DESCRIPTION)
- src/chronulus_mcp/__init__.py:7-7 (registration)Imports the create_prediction_agent_and_get_predictions function from the predictor module to make it available for registration.from chronulus_mcp.agent.predictor import create_prediction_agent_and_get_predictions, reuse_prediction_agent_and_get_prediction
- Pydantic schema definitions for the tool's input parameters using Annotated types with Field descriptions.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]]]: