vac_routes.py

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

Classes

VACRequest

Request model for VAC endpoints.

• copy(self) -> 'Self'

  • Returns a shallow copy of the model.

• deepcopy(self, memo: 'dict[int, Any] | None' = None) -> 'Self'

  • Returns a deep copy of the model.

• delattr(self, item: 'str') -> 'Any'

  • Implement delattr(self, name).

• eq(self, other: 'Any') -> 'bool'

  • Return self==value.

• getattr(self, item: 'str') -> 'Any'

  • No docstring available.

• getstate(self) -> 'dict[Any, Any]'

  • Helper for pickle.

• init(self, /, **data: 'Any') -> 'None'

  • Create a new model by parsing and validating input data from keyword arguments.

Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

`self` is explicitly positional-only to allow `self` as a field name.

• iter(self) -> 'TupleGenerator'

  • So `dict(model)` works.

• pretty(self, fmt: 'typing.Callable[[Any], Any]', **kwargs: 'Any') -> 'typing.Generator[Any, None, None]'

  • Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

• replace(self, **changes: 'Any') -> 'Self'

  • No docstring available.

• repr(self) -> 'str'

  • Return repr(self).

• repr_args(self) -> '_repr.ReprArgs'

  • No docstring available.

• repr_name(self) -> 'str'

  • Name of the instance's class, used in repr.

• repr_recursion(self, object: 'Any') -> 'str'

  • Returns the string representation of a recursive object.

• repr_str(self, join_str: 'str') -> 'str'

  • No docstring available.

• rich_repr(self) -> 'RichReprResult'

  • Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

• setattr(self, name: 'str', value: 'Any') -> 'None'

  • Implement setattr(self, name, value).

• setstate(self, state: 'dict[Any, Any]') -> 'None'

  • No docstring available.

• str(self) -> 'str'

  • Return str(self).

• _calculate_keys(self, *args: 'Any', **kwargs: 'Any') -> 'Any'

  • No docstring available.

• _copy_and_set_values(self, *args: 'Any', **kwargs: 'Any') -> 'Any'

  • No docstring available.

• _iter(self, *args: 'Any', **kwargs: 'Any') -> 'Any'

  • No docstring available.

• _setattr_handler(self, name: 'str', value: 'Any') -> 'Callable[[BaseModel, str, Any], None] | None'

  • Get a handler for setting an attribute on the model instance.

Returns: A handler for setting an attribute on the model instance. Used for memoization of the handler. Memoizing the handlers leads to a dramatic performance improvement in `setattr` Returns `None` when memoization is not safe, then the attribute is set directly.

• copy(self, *, include: 'AbstractSetIntStr | MappingIntStrAny | None' = None, exclude: 'AbstractSetIntStr | MappingIntStrAny | None' = None, update: 'Dict[str, Any] | None' = None, deep: 'bool' = False) -> 'Self'
  • Returns a copy of the model.

!!! warning "Deprecated" This method is now deprecated; use model_copy instead.

If you need include or exclude, use:

data = self.model_dump(include=include, exclude=exclude, round_trip=True)
data = {**data, **(update or {})}
copied = self.model_validate(data)

Args: include: Optional set or mapping specifying which fields to include in the copied model. exclude: Optional set or mapping specifying which fields to exclude in the copied model. update: Optional dictionary of field-value pairs to override field values in the copied model. deep: If True, the values of fields that are Pydantic models will be deep-copied.

Returns: A copy of the model with included, excluded and updated fields as specified.

• dict(self, *, include: 'IncEx | None' = None, exclude: 'IncEx | None' = None, by_alias: 'bool' = False, exclude_unset: 'bool' = False, exclude_defaults: 'bool' = False, exclude_none: 'bool' = False) -> 'Dict[str, Any]'

  • No docstring available.

• json(self, *, include: 'IncEx | None' = None, exclude: 'IncEx | None' = None, by_alias: 'bool' = False, exclude_unset: 'bool' = False, exclude_defaults: 'bool' = False, exclude_none: 'bool' = False, encoder: 'Callable[[Any], Any] | None' = PydanticUndefined, models_as_dict: 'bool' = PydanticUndefined, **dumps_kwargs: 'Any') -> 'str'

  • No docstring available.

• model_copy(self, *, update: 'Mapping[str, Any] | None' = None, deep: 'bool' = False) -> 'Self'

  • !!! abstract "Usage Documentation" `model_copy`

Returns a copy of the model.

!!! note The underlying instance's [`dict`][object.dict] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Args: update: Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data. deep: Set to `True` to make a deep copy of the model.

Returns: New model instance.

• model_dump(self, *, mode: "Literal['json', 'python'] | str" = 'python', include: 'IncEx | None' = None, exclude: 'IncEx | None' = None, context: 'Any | None' = None, by_alias: 'bool | None' = None, exclude_unset: 'bool' = False, exclude_defaults: 'bool' = False, exclude_none: 'bool' = False, round_trip: 'bool' = False, warnings: "bool | Literal['none', 'warn', 'error']" = True, fallback: 'Callable[[Any], Any] | None' = None, serialize_as_any: 'bool' = False) -> 'dict[str, Any]'
  • !!! abstract "Usage Documentation" `model_dump`

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Args: mode: The mode in which `to_python` should run. If mode is 'json', the output will only contain JSON serializable types. If mode is 'python', the output may contain non-JSON-serializable Python objects. include: A set of fields to include in the output. exclude: A set of fields to exclude from the output. context: Additional context to pass to the serializer. by_alias: Whether to use the field's alias in the dictionary key if defined. exclude_unset: Whether to exclude fields that have not been explicitly set. exclude_defaults: Whether to exclude fields that are set to their default value. exclude_none: Whether to exclude fields that have a value of `None`. round_trip: If True, dumped values should be valid as input for non-idempotent types such as Json[T]. warnings: How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, "error" raises a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError]. fallback: A function to call when an unknown value is encountered. If not provided, a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError] error is raised. serialize_as_any: Whether to serialize fields with duck-typing serialization behavior.

Returns: A dictionary representation of the model.

• model_dump_json(self, *, indent: 'int | None' = None, include: 'IncEx | None' = None, exclude: 'IncEx | None' = None, context: 'Any | None' = None, by_alias: 'bool | None' = None, exclude_unset: 'bool' = False, exclude_defaults: 'bool' = False, exclude_none: 'bool' = False, round_trip: 'bool' = False, warnings: "bool | Literal['none', 'warn', 'error']" = True, fallback: 'Callable[[Any], Any] | None' = None, serialize_as_any: 'bool' = False) -> 'str'
  • !!! abstract "Usage Documentation" `model_dump_json`

Generates a JSON representation of the model using Pydantic's `to_json` method.

Args: indent: Indentation to use in the JSON output. If None is passed, the output will be compact. include: Field(s) to include in the JSON output. exclude: Field(s) to exclude from the JSON output. context: Additional context to pass to the serializer. by_alias: Whether to serialize using field aliases. exclude_unset: Whether to exclude fields that have not been explicitly set. exclude_defaults: Whether to exclude fields that are set to their default value. exclude_none: Whether to exclude fields that have a value of `None`. round_trip: If True, dumped values should be valid as input for non-idempotent types such as Json[T]. warnings: How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, "error" raises a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError]. fallback: A function to call when an unknown value is encountered. If not provided, a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError] error is raised. serialize_as_any: Whether to serialize fields with duck-typing serialization behavior.

Returns: A JSON string representation of the model.

• model_post_init(self, context: 'Any', /) -> 'None'
  • Override this method to perform additional initialization after `init` and `model_construct`. This is useful if you want to do some validation that requires the entire model to be initialized.

VACRoutesFastAPI

FastAPI implementation of VAC routes with streaming support and extensible MCP integration.

This class provides a comprehensive FastAPI application with:

• VAC (Virtual Agent Computer) endpoints for AI chat and streaming
• OpenAI-compatible API endpoints
• Extensible MCP (Model Context Protocol) server integration for Claude Desktop/Code
• MCP client support for connecting to external MCP servers
• A2A (Agent-to-Agent) protocol support
• Server-Sent Events (SSE) streaming capabilities

Key Features

1. VAC Endpoints

• /vac/{vector_name} - Non-streaming VAC responses
• /vac/streaming/{vector_name} - Plain text streaming responses
• /vac/streaming/{vector_name}/sse - Server-Sent Events streaming

2. OpenAI Compatible API

• /openai/v1/chat/completions - OpenAI-compatible chat completions
• Supports both streaming and non-streaming modes

3. MCP Integration

• MCP Server: Expose your VAC as MCP tools for Claude Desktop/Code
• MCP Client: Connect to external MCP servers and use their tools
• Custom Tools: Easily add your own MCP tools using decorators

4. A2A Agent Protocol

• Agent discovery and task execution
• Compatible with multi-agent workflows

Basic Usage

Simplified Setup (Recommended)

Use the helper method for automatic lifespan management:

from sunholo.agents.fastapi import VACRoutesFastAPI

async def my_stream_interpreter(question, vector_name, chat_history, callback, **kwargs):
    # Your streaming VAC logic here
    # Use callback.async_on_llm_new_token(token) for streaming
    return {"answer": "Response", "sources": []}

# Single call sets up everything with MCP server and proper lifespan management
app, vac_routes = VACRoutesFastAPI.create_app_with_mcp(
    title="My VAC Application",
    stream_interpreter=my_stream_interpreter
    # MCP server is automatically enabled when using this method
)

# Add custom endpoints if needed
@app.get("/custom")
async def custom_endpoint():
    return {"message": "Hello"}

# Run the app
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Manual Setup (Advanced)

For more control over lifespan management:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from sunholo.agents.fastapi import VACRoutesFastAPI

async def my_stream_interpreter(question, vector_name, chat_history, callback, **kwargs):
    return {"answer": "Response", "sources": []}

# Define your app's lifespan
@asynccontextmanager
async def app_lifespan(app: FastAPI):
    print("Starting up...")
    yield
    print("Shutting down...")

# Create temp app to get MCP lifespan
temp_app = FastAPI()
vac_routes_temp = VACRoutesFastAPI(
    temp_app,
    stream_interpreter=my_stream_interpreter,
    enable_mcp_server=True
)

# Get MCP lifespan
mcp_lifespan = vac_routes_temp.get_mcp_lifespan()

# Combine lifespans
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
    async with app_lifespan(app):
        if mcp_lifespan:
            async with mcp_lifespan(app):
                yield
        else:
            yield

# Create app with combined lifespan
app = FastAPI(title="My VAC Application", lifespan=combined_lifespan)

# Initialize VAC routes
vac_routes = VACRoutesFastAPI(
    app=app,
    stream_interpreter=my_stream_interpreter,
    enable_mcp_server=True
)

Your FastAPI app now includes:

• All VAC endpoints
• MCP server at /mcp (for Claude Desktop/Code to connect)
• Built-in VAC tools: vac_stream, vac_query, list_available_vacs, get_vac_info

Adding Custom MCP Tools

Method 1: Using Decorators

vac_routes = VACRoutesFastAPI(app, stream_interpreter, enable_mcp_server=True)

@vac_routes.add_mcp_tool
async def get_weather(city: str) -> str:
    '''Get weather information for a city.'''
    # Your weather API logic
    return f"Weather in {city}: Sunny, 22°C"

@vac_routes.add_mcp_tool("custom_search", "Search our database")
async def search_database(query: str, limit: int = 10) -> list:
    '''Search internal database with custom name and description.'''
    # Your database search logic
    return [{"result": f"Found: {query}"}]

Method 2: Programmatic Registration

async def my_business_tool(param: str) -> dict:
    return {"processed": param}

# Add tool with custom name and description
vac_routes.add_mcp_tool(
    my_business_tool, 
    "process_business_data", 
    "Process business data with our custom logic"
)

Method 3: Advanced MCP Server Access

# Get direct access to MCP server for advanced customization
mcp_server = vac_routes.get_mcp_server()

@mcp_server.add_tool
async def advanced_tool(complex_param: dict) -> str:
    return f"Advanced processing: {complex_param}"

# List all registered tools
print("Available MCP tools:", vac_routes.list_mcp_tools())

MCP Client Integration

Connect to external MCP servers and use their tools:

mcp_servers = [
    {
        "name": "filesystem-server",
        "command": "npx",
        "args": ["@modelcontextprotocol/server-filesystem", "/path/to/files"]
    }
]

vac_routes = VACRoutesFastAPI(
    app, stream_interpreter,
    mcp_servers=mcp_servers,  # Connect to external MCP servers
    enable_mcp_server=True    # Also expose our own MCP server
)

# External MCP tools available at:
# GET /mcp/tools - List all external tools
# POST /mcp/call - Call external MCP tools

Claude Desktop Integration

Option 1: Remote Integration (Recommended for Development)

# Run your FastAPI app
uvicorn.run(vac_routes.app, host="0.0.0.0", port=8000)

# Configure Claude Desktop (Settings > Connectors > Add custom connector):
# URL: http://localhost:8000/mcp

Option 2: Local Integration

Create a standalone script for Claude Desktop:

# claude_mcp_server.py
from sunholo.mcp.extensible_mcp_server import create_mcp_server

server = create_mcp_server("my-app", include_vac_tools=True)

@server.add_tool
async def my_app_tool(param: str) -> str:
    return f"My app processed: {param}"

if __name__ == "__main__":
    server.run()

# Install: fastmcp install claude-desktop claude_mcp_server.py --with sunholo[anthropic]

Available Built-in MCP Tools

When enable_mcp_server=True, these tools are automatically available:

• vac_stream: Stream responses from any configured VAC
• vac_query: Query VACs with non-streaming responses
• list_available_vacs: List all available VAC configurations
• get_vac_info: Get detailed information about a specific VAC

Error Handling and Best Practices

@vac_routes.add_mcp_tool
async def robust_tool(user_input: str) -> str:
    '''Example of robust tool implementation.'''
    try:
        # Validate input
        if not user_input or len(user_input) > 1000:
            return "Error: Invalid input length"
        
        # Your business logic
        result = await process_user_input(user_input)
        
        return f"Processed: {result}"
        
    except Exception as e:
        # Log error and return user-friendly message
        log.error(f"Tool error: {e}")
        return f"Error processing request: {str(e)}"

Configuration Options

vac_routes = VACRoutesFastAPI(
    app=app,
    stream_interpreter=my_stream_func,
    vac_interpreter=my_vac_func,           # Optional non-streaming function
    additional_routes=[],                   # Custom FastAPI routes
    mcp_servers=[],                        # External MCP servers to connect to
    add_langfuse_eval=True,                # Enable Langfuse evaluation
    enable_mcp_server=True,                # Enable MCP server for Claude
    enable_a2a_agent=False,                # Enable A2A agent protocol
    a2a_vac_names=None                     # VACs available for A2A
)

• init(self, app: 'FastAPI', stream_interpreter: 'Callable', vac_interpreter: 'Optional[Callable]' = None, additional_routes: 'Optional[List[Dict]]' = None, mcp_servers: 'Optional[List[Dict[str, Any]]]' = None, add_langfuse_eval: 'bool' = True, enable_a2a_agent: 'bool' = False, a2a_vac_names: 'Optional[List[str]]' = None)
  • Initialize FastAPI VAC routes with comprehensive AI and MCP integration.

Args: app: FastAPI application instance to register routes on stream_interpreter: Function for streaming VAC responses. Can be async or sync. Called with (question, vector_name, chat_history, callback, **kwargs) vac_interpreter: Optional function for non-streaming VAC responses. If not provided, will use stream_interpreter without streaming callbacks. additional_routes: List of custom route dictionaries to register: [{"path": "/custom", "handler": func, "methods": ["GET"]}] mcp_servers: List of external MCP server configurations to connect to: [{"name": "server-name", "command": "python", "args": ["server.py"]}] add_langfuse_eval: Whether to enable Langfuse evaluation and tracing enable_a2a_agent: Whether to enable A2A (Agent-to-Agent) protocol endpoints a2a_vac_names: List of VAC names available for A2A agent interactions

Stream Interpreter Function

Your stream_interpreter should handle streaming responses:

async def my_stream_interpreter(question: str, vector_name: str, 
                              chat_history: list, callback, **kwargs):
    # Process the question using your AI/RAG pipeline
    
    # For streaming tokens:
    await callback.async_on_llm_new_token("partial response...")
    
    # Return final result with sources:
    return {
        "answer": "Final complete answer",
        "sources": [{"title": "Source 1", "url": "..."}]
    }

MCP Server Integration

When VACMCPServer is available, the following happens automatically:

1. MCP server is mounted at /mcp/mcp endpoint (NOTE: /mcp/mcp not /mcp!)
2. Built-in VAC tools are automatically registered:
   • vac_stream, vac_query, list_available_vacs, get_vac_info
3. You can add custom MCP tools using add_mcp_tool()
4. Claude Desktop/Code can connect to http://your-server/mcp/mcp

IMPORTANT: The endpoint is /mcp/mcp to avoid the MCP app intercepting other routes. DO NOT change the mounting point to "" (root) as it will break other FastAPI routes!

Complete Example

app = FastAPI(title="My VAC Application")

async def my_vac_logic(question, vector_name, chat_history, callback, **kwargs):
    # Your AI/RAG implementation
    result = await process_with_ai(question)
    return {"answer": result, "sources": []}

# External MCP servers to connect to
external_mcp = [
    {"name": "filesystem", "command": "mcp-server-fs", "args": ["/data"]}
]

vac_routes = VACRoutesFastAPI(
    app=app,
    stream_interpreter=my_vac_logic,
    mcp_servers=external_mcp,
    enable_mcp_server=True  # Enable for Claude integration
)

# Add custom MCP tools for your business logic
@vac_routes.add_mcp_tool
async def get_customer_info(customer_id: str) -> dict:
    return await fetch_customer(customer_id)

# Your app now has:
# - VAC endpoints: /vac/{vector_name}, /vac/streaming/{vector_name}
# - OpenAI API: /openai/v1/chat/completions
# - MCP server: /mcp (with built-in + custom tools)
# - MCP client: /mcp/tools, /mcp/call (for external servers)

• _get_or_create_a2a_agent(self, request: 'Request')

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

• _initialize_mcp_servers(self)

  • Initialize connections to configured MCP servers.

• _register_custom_tools(self)

  • Register any custom tools that were added before MCP server initialization.

• _setup_lifespan(self)

  • Set up lifespan context manager for app initialization.

• add_mcp_resource(self, func: 'Callable', name: 'str' = None, description: 'str' = None)

  • Add a custom MCP resource to the server.

Args: func: The resource function name: Optional custom name for the resource description: Optional description (uses docstring if not provided)

Example: @app.add_mcp_resource async def my_custom_resource(uri: str) -> str: '''Custom resource that provides data.''' return f"Resource data for: {uri}"

• add_mcp_tool(self, func: 'Callable', name: 'str' = None, description: 'str' = None)
  • Add a custom MCP tool to the server.

Args: func: The tool function name: Optional custom name for the tool description: Optional description (uses docstring if not provided)

Example: @app.add_mcp_tool async def my_custom_tool(param: str) -> str: '''Custom tool that does something useful.''' return f"Result: {param}"

Or with custom name and description

app.add_mcp_tool(my_function, "custom_name", "Custom description")

• create_app_with_mcp(title: 'str' = 'VAC Application', stream_interpreter: 'Optional[callable]' = None, vac_interpreter: 'Optional[callable]' = None, app_lifespan: 'Optional[callable]' = None, **kwargs) -> "tuple[FastAPI, 'VACRoutesFastAPI']"
  • Helper method to create a FastAPI app with proper MCP lifespan management.

This method simplifies the setup process by handling the lifespan combination automatically, avoiding the need for the double initialization pattern. MCP server is automatically enabled when using this method.

Args: title: Title for the FastAPI app stream_interpreter: Streaming interpreter function vac_interpreter: Non-streaming interpreter function
app_lifespan: Optional app lifespan context manager **kwargs: Additional arguments passed to VACRoutesFastAPI

Returns: Tuple of (FastAPI app, VACRoutesFastAPI instance)

Example:

from sunholo.agents.fastapi import VACRoutesFastAPI

async def my_interpreter(question, vector_name, chat_history, callback, **kwargs):
    # Your logic here
    return {"answer": "response", "sources": []}

# Single call to set up everything (MCP is automatically enabled)
app, vac_routes = VACRoutesFastAPI.create_app_with_mcp(
    title="My VAC App",
    stream_interpreter=my_interpreter
)

# Add custom endpoints
@app.get("/custom")
async def custom_endpoint():
    return {"message": "Custom endpoint"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

• get_mcp_lifespan(self)
  • Get the MCP app's lifespan for manual lifespan management.

Returns: The MCP app's lifespan if MCP server is enabled, None otherwise.

Example:

from contextlib import asynccontextmanager

# Create temp app to get MCP lifespan
temp_app = FastAPI()
vac_routes = VACRoutesFastAPI(temp_app, ..., enable_mcp_server=True)
mcp_lifespan = vac_routes.get_mcp_lifespan()

# Combine with your app's lifespan
@asynccontextmanager
async def combined_lifespan(app: FastAPI):
    async with my_app_lifespan(app):
        if mcp_lifespan:
            async with mcp_lifespan(app):
                yield
        else:
            yield

app = FastAPI(lifespan=combined_lifespan)

• get_mcp_server(self)
  • Get the MCP server instance for advanced customization.

Returns: VACMCPServer instance or None if MCP server is not enabled

• handle_a2a_agent_card(self, request: 'Request')

  • Handle A2A agent card discovery request.

• handle_a2a_push_notification(self, request: 'Request')

  • Handle A2A push notification settings.

• handle_a2a_task_cancel(self, request: 'Request')

  • Handle A2A task cancel request.

• handle_a2a_task_get(self, request: 'Request')

  • Handle A2A task get request.

• handle_a2a_task_send(self, request: 'Request')

  • Handle A2A task send request.

• handle_a2a_task_send_subscribe(self, request: 'Request')

  • Handle A2A task send with subscription (SSE).

• handle_mcp_call_tool(self, request: 'Request')

  • Call an MCP tool.

• handle_mcp_list_resources(self, request: 'Request')

  • List available MCP resources.

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

  • List available MCP tools.

• handle_mcp_read_resource(self, request: 'Request')

  • Read an MCP resource.

• handle_openai_compatible(self, request: 'Request', vector_name: 'Optional[str]' = None)

  • Handle OpenAI-compatible chat completion requests.

• handle_process_vac(self, vector_name: 'str', request: 'Request')

  • Handle non-streaming VAC requests.

• handle_stream_vac(self, vector_name: 'str', request: 'Request')

  • Handle streaming VAC requests with plain text response. Compatible with Flask implementation.

• handle_stream_vac_sse(self, vector_name: 'str', request: 'Request')

  • Handle streaming VAC requests with Server-Sent Events format. Better for browser-based clients.

• health(self)

  • Health check endpoint.

• home(self)

  • Home endpoint.

• list_mcp_resources(self) -> 'List[str]'

  • List all registered MCP resources.

Returns: List of resource names

• list_mcp_tools(self) -> 'List[str]'
  • List all registered MCP tools.

Returns: List of tool names

• openai_health(self)

  • OpenAI health check endpoint.

• prep_vac_async(self, vac_request: 'VACRequest', vector_name: 'str')

  • Prepare VAC request data asynchronously.

• register_routes(self)

  • Register all VAC routes with the FastAPI application.

• vac_interpreter_default(self, question: 'str', vector_name: 'str', chat_history=None, **kwargs)

  • Default VAC interpreter that uses the stream interpreter without streaming.