vac_routes.py

Source: src/sunholo/agents/flask/vac_routes.py

Classes

VACRoutes

Usage Example:

from agents.flask import VACRoutes

app = Flask(__name__)

def stream_interpreter(question, vector_name, chat_history, **kwargs):
    # Implement your streaming logic
    ...

def vac_interpreter(question, vector_name, chat_history, **kwargs):
    # Implement your static VAC logic
    ...

vac_routes = VACRoutes(app, stream_interpreter, vac_interpreter)

if __name__ == "__main__":
    app.run(debug=True)

• init(self, app, stream_interpreter: <built-in function callable>, vac_interpreter: <built-in function callable> = None, additional_routes: dict = None, mcp_servers: List[Dict[str, Any]] = None, async_stream: bool = False, add_langfuse_eval: bool = True, enable_mcp_server: bool = False, enable_a2a_agent: bool = False, a2a_vac_names: List[str] = None)

  • Initialize self. See help(type(self)) for accurate signature.

• _async_generator_to_stream(async_gen_func)

  • Helper function to stream the async generator's values to the client.

• _get_or_create_a2a_agent(self)

  • Get or create the A2A agent instance with current request context.

• _initialize_mcp_servers(self)

  • Initialize connections to configured MCP servers.

• check_authentication(self)

  • No docstring available.

• create_langfuse_trace(self, request, vector_name, trace_id)

  • No docstring available.

• handle_a2a_agent_card(self)

  • Handle A2A agent card discovery request.

• handle_a2a_push_notification(self)

  • Handle A2A push notification settings.

• handle_a2a_task_cancel(self)

  • Handle A2A task cancel request.

• handle_a2a_task_get(self)

  • Handle A2A task get request.

• handle_a2a_task_send(self)

  • Handle A2A task send request.

• handle_a2a_task_send_subscribe(self)

  • Handle A2A task send with subscription (SSE).

• handle_file_upload(self, file, vector_name)

  • No docstring available.

• handle_mcp_call_tool(self)

  • Call an MCP tool.

• handle_mcp_list_resources(self)

  • List available MCP resources.

• handle_mcp_list_tools(self, server_name: Optional[str] = None)

  • List available MCP tools.

• handle_mcp_read_resource(self)

  • Read an MCP resource.

• handle_mcp_server(self)

  • Handle MCP server requests using HTTP transport.

• handle_openai_compatible_endpoint(self, vector_name=None)

  • No docstring available.

• handle_options(self, **kwargs)

  • No docstring available.

• handle_process_vac(self, vector_name)

  • No docstring available.

• handle_stream_vac(self, vector_name)

  • No docstring available.

• handle_stream_vac_async(self, vector_name)

  • No docstring available.

• health(self)

  • No docstring available.

• home(self)

  • No docstring available.

• langfuse_eval_response(self, trace_id, eval_percent=0.01)

  • Sends an evaluation message based on a probability defined by eval_percent.

Args: eval_percent (float): The probability (0 to 1) of triggering the evaluation. trace_id (str): The trace identifier for the evaluation.

Returns: None

• make_openai_response(self, user_message, vector_name, answer)

  • No docstring available.

• openai_health_endpoint()

  • No docstring available.

• prep_vac(self, request, vector_name)

  • No docstring available.

• prep_vac_async(self, request, vector_name)

  • Async version of prep_vac.

• register_additional_routes(self)

  • Registers additional custom routes provided during initialization.

Example:

from flask import Flask, jsonify
from agents.flask import VACRoutes

app = Flask(__name__)

def stream_interpreter(question, vector_name, chat_history, **kwargs):
    # Implement your streaming logic
    ...

def vac_interpreter(question, vector_name, chat_history, **kwargs):
    # Implement your static VAC logic
    ...

def custom_handler():
    return jsonify({"message": "Custom route!"})

custom_routes = [
    {
        "rule": "/custom",
        "methods": ["GET"],
        "handler": custom_handler
    }
]

vac_routes = VACRoutes(app, stream_interpreter, vac_interpreter, additional_routes=custom_routes)

if __name__ == "__main__":
    app.run(debug=True)

• register_after_request(self, response)

  • No docstring available.

• register_routes(self)

  • Registers all the VAC routes for the Flask application.

• vac_interpreter_default(self, question: str, vector_name: str, chat_history=[], **kwargs)

  • No docstring available.