vs-agent

Full-stack agent framework for Viveka Sutra — build agents that serve over HTTP, MCP, queues, or any combination, from a single codebase.

---

Overview

vs-agent combines vs-server (HTTP/WebSocket/SSE) and vs-mcp-agent (MCP) into a single VsAgentServer. It lets you write a capability once with @vs_agent_action and declare, per transport, where it can be reached from — an HTTP route, an MCP tool, a queue topic, or any combination — with the same auth, guards, and business logic.

Each transport is opt-in by being present. No http block means no route; no mcp block means no model ever sees it. The way to not expose something is to not mention it, rather than to remember a flag.

The library is fully server-agnostic. VsAgentServer does not hardcode FastAPI or FastMCP — it uses VsServerFactory and VsMcpServerFactory to resolve the correct server implementation at startup. Swapping or adding a new server type requires no changes to application code.

---

The Problem It Solves

An agent needs to be callable by both humans (via HTTP) and AI clients (via MCP). Without vs-agent, you register the same function twice, apply guards twice, and maintain two sets of route definitions.

Without vs-agent

# HTTP route
@router.post("/v1/docs/search")
async def search_docs_http(query: str, auth=Depends(require_auth)):
    return await _search(query)

# MCP tool — separate registration, separate guard wiring
@mcp.tool(name="search_docs")
async def search_docs_mcp(query: str) -> dict:
    await require_auth()
    return await _search(query)

With vs-agent

@vs_agent_action(
    name="search_docs",
    description="Search VS library documentation",
    http={"url": "/v1/docs/search", "method": "POST"},
    mcp={"name": "search_docs"},
    guards=[VsActionSecurity(roles=["user"])],
)
async def search_docs(query: str) -> VsToolResponse:
    return VsToolResponse(status="success", result=await _search(query))

One function, one guard, available on both protocols — and reachable only over the protocols it names.

---

Installation

pip install vs-agent

With MCP support:

pip install vs-agent[mcp]

With auth guard support:

pip install vs-agent[mcp,security]

---

Dependencies

Library	Required	Purpose
vs-common	Yes	Config, logging
vs-server	Yes	HTTP/WebSocket/SSE server
vs-mcp-agent	No — install with [mcp] extra	MCP server
vs-security	No — install with [security] extra	JWT auth and VsActionSecurity guard
vs-queue	Optional — [queue] extra	Queue request/reply transport (queue={"listen": ...})

---

Configuration

VsAgentServer reads HTTP config via vs-server and MCP config via vs-mcp-agent. Both use the same config.ini file — separated by section.

HTTP section (from vs-server):

Key	Default	Description
server.name	vs-agent	Server name shown in logs and / response
server.version	0.1.0	Version shown in logs and / response
server.host	0.0.0.0	Host to bind
server.port	8000	HTTP port
server.reload	false	Enable hot reload (dev only)
server.workers	1	Number of worker processes
server.cors_origins	*	Comma-separated allowed CORS origins
server.ssl_certfile	—	Path to TLS certificate file
server.ssl_keyfile	—	Path to TLS private key file

MCP section (from vs-mcp-agent):

Key	Default	Description
agent.name	vs-agent	MCP server name
agent.version	0.1.0	MCP server version
agent.transport	streamable-http	Transport: stdio, sse, streamable-http
agent.host	0.0.0.0	MCP host
agent.port	8080	MCP port
agent.enable_auth	false	Enable authentication for HTTP requests and MCP HTTP transports

Authentication is disabled when agent.enable_auth is absent or false. This replaces the former agent.auth_enabled key; migrate existing configuration explicitly. When enabled, pass a VsRequestAuthenticator instance to VsAgentServer. The same instance authenticates HTTP and MCP requests; no auth manager or implicit JWT provider is created. Initialization errors are logged and propagated. Action guards still run when authentication is disabled. MCP stdio calls the same authenticator before message dispatch; credentials must come from an explicit non-HTTP source.

config.ini example:

[server]
name = my-agent
version = 1.0.0
host = 0.0.0.0
port = 8000

[agent]
name = my-agent
version = 1.0.0
transport = streamable-http
host = 0.0.0.0
port = 8080
enable_auth = true

[auth]
secret_key = your-secret-key
algorithm = HS256

[logging]
level = INFO
file_path = ./logs/agent.log

---

Quick Start

HTTP + MCP (most common)

from vs_common.config.vs_ini_config import VsIniConfig
from vs_common.log.vs_log_manager import VsLogManager
from vs_common.schema.vs_log_config import VsLogConfig
from vs_server.server.vs_fast_api_server import VsFastApiServer  # noqa — auto-registers "fastapi"
from vs_mcp_agent.server.vs_fast_mcp_server import VsFastMcpServer  # noqa — auto-registers "fastmcp"
from vs_agent.server.vs_agent_server import VsAgentServer

import my_agent.actions  # noqa — registers @vs_agent_action functions


def main():
    config = VsIniConfig("config.ini")
    VsLogManager.init(VsLogConfig(level=config.get("logging.level", default="INFO")))

    server = VsAgentServer(config, http="fastapi", mcp="fastmcp")
    server.add_actions()
    server.run()


if __name__ == "__main__":
    main()

HTTP only

server = VsAgentServer(config, http="fastapi")
server.add_actions()
server.run()

MCP only

from vs_mcp_agent.server.vs_fast_mcp_server import VsFastMcpServer  # noqa — auto-registers "fastmcp"

server = VsAgentServer(config, mcp="fastmcp")
server.add_actions()   # required — this is what hands MCP-declaring actions to the tool registry
server.run()

---

How It All Fits Together

Application Startup
    └── import VsFastApiServer   # auto-registers "fastapi" into VsServerFactory
    └── import VsFastMcpServer   # auto-registers "fastmcp" into VsMcpServerFactory
    └── import my_agent.actions  # @vs_agent_action decorators self-register into VsActionRegistry

VsAgentServer(config, http="fastapi", mcp="fastmcp")
    ├── VsServerFactory.get("fastapi", config)      → VsFastApiServer
    └── VsMcpServerFactory.get("fastmcp", config)   → VsFastMcpServer

server.add_actions()
    ├── VsActionRegistry.http_actions()  → HTTP routes + /capabilities
    ├── VsActionRegistry.mcp_actions()   → VsToolRegistry (VsFastMcpServer wires it at run())
    └── VsActionRegistry.queue_actions() → VsQueueActionConsumer → VsQueueManager → vs-queue

server.run()
    ├── MCP server starts on background thread (port 8080)
    └── HTTP server starts on main thread (port 8000)

---

VsAgentServer

VsAgentServer is the central coordinator. It creates and manages HTTP and MCP server instances via their respective factories.

from vs_agent.server.vs_agent_server import VsAgentServer

server = VsAgentServer(config, http="fastapi", mcp="fastmcp")

Constructor parameters:

Parameter	Type	Default	Description
config	VsBaseConfig	—	Application config
http	Optional[str]	"fastapi"	HTTP server key. Pass None for MCP-only mode.
mcp	Optional[str]	None	MCP server key. Pass "fastmcp" to enable MCP.
authenticator	Optional[VsRequestAuthenticator]	None	Shared authentication instance; required when agent.enable_auth=true.
action_modules	Optional[Sequence[str]]	None	Dotted action modules or packages to import; packages are scanned recursively.
queue	Optional[VsBaseQueue]	None	Disconnected vs-queue provider instance; connected and managed by the queue runtime
queue_reply_guard	Callable	None	Authorizes (VsAuthRequest, AuthContext) reply destinations; required for authenticated queue agents

At least one of http or mcp must be specified — both None require a queue instance; otherwise construction raises ValueError.

Methods:

Method	Description
add_controller(controller)	Register an HTTP @controller class. Delegates to the HTTP server.
add_router(router)	Register a raw router. Delegates to the HTTP server.
add_websocket(handler)	Register a @websocket handler. Delegates to the HTTP server.
add_sse(handler)	Register an @sse handler. Delegates to the HTTP server.
add_actions()	Wire every registered action to the transports it declares — HTTP routes plus /capabilities, and MCP tools.
get_app()	Return the underlying ASGI app (e.g. FastAPI instance).
run()	Start the server(s). MCP runs on a background daemon thread; HTTP runs on the main thread.

mcp property:

server.mcp  # returns the underlying FastMCP instance for advanced configuration

Raises RuntimeError if no MCP server is configured.

All add_* methods return self for chaining:

server = (
    VsAgentServer(config, http="fastapi", mcp="fastmcp")
    .add_controller(HealthController())
    .add_actions()
)
server.run()

---

@vs_agent_action

Registers a function as one action, and declares per transport where it can be reached from.

from vs_agent.decorator.vs_agent_action_decorator import vs_agent_action
from vs_agent.auth.vs_action_security import VsActionSecurity
from vs_mcp_agent.schema.vs_tool_response import VsToolResponse

@vs_agent_action(
    name="ag.create_new_agent",
    description="Create a new Viveka Sutra agent with the given name and description",
    http={"url": "/v1/agent", "method": "POST"},
    mcp={"name": "create_new_agent"},
    queue={"listen": "agent.create"},
    guards=[VsActionSecurity(roles=["admin"])],
)
async def create_new_agent(request: CreateAgentRequest) -> VsToolResponse:
    ...

Parameters:

Parameter	Type	Required	Default	Description
name	str	Yes	—	Action identifier. Must be unique; duplicates raise ValueError
description	str	Yes	—	Shown in /capabilities and, unless overridden, in the MCP tool list
http	dict / HttpTransport	No	None	Present → served over HTTP. {"url": ..., "method": "POST"}
mcp	dict / McpTransport	No	None	Present → exposed as an MCP tool. {} accepts all defaults
queue	dict / QueueTransport	No	None	Present → consumes a queue topic. {"listen": "topic"}
intents	List[Intent]	No	[]	Semantic intents for orchestrator routing
input_schema	Dict[str, Any]	No	None	JSON schema for the action input
guards	List[Callable]	No	[]	Run before the function on every transport
capability	bool	No	True	Whether to advertise the action at all

At least one transport must be declared — an action with none raises ValueError.

Registration also rejects contradictory capability=False plus mcp declarations, unknown fields in any transport, invalid HTTP methods, and HTTP paths without a leading /. HTTP methods must be uppercase: GET, POST, PUT, PATCH, DELETE, HEAD, OPTIONS, or TRACE. The path alias for HTTP url remains supported.

Two actions cannot declare the same HTTP URL and method or resolve to the same MCP tool name. Conflict errors name both actions. These checks happen during decoration/registration, before server wiring. HTTP-only management actions with capability=False remain valid; omit mcp for those actions.

Synchronous and asynchronous actions are supported over HTTP and MCP. Guarded MCP synchronous actions and synchronous guards run in a worker thread with the authenticated context available. Guards must not require HTTP-specific arguments to work on both transports. Queue actions use the same named-parameter contract and support synchronous/asynchronous functions and guards.

The registry copies the supplied guard list, intents, input schema, and transport settings. Changing your original lists or metadata after decoration does not alter the registered action.

Transports are opt-in by presence

Declaration	Result
http={"url": "/v1/agent"} only	HTTP route. No MCP tool, no queue consumer
mcp={} only	MCP tool named after the action. No route at all
queue={"listen": "agent.create"} only	Queue consumer. Nothing reachable over HTTP or MCP
all three	One function, three ways in

http.method defaults to POST. http also accepts path in place of url, so an @action(path=...) reads the same when migrated. mcp.name defaults to the action name, and mcp.description to the action description — override either when the name a model reads should differ from the internal action id.

If an action declares a transport whose server was not started (mcp={...} with no mcp= passed to VsAgentServer), that is not an error — add_actions() logs a warning and the action is simply unreachable that way.

guards are the action's, not a transport's

One guards list runs before the function no matter which transport the call arrived over. There is deliberately no per-transport guard list: an action that is safe to reach one way and not another is two actions, not one with a clever guard list.

capability is a separate question

A transport says how an action can be reached. capability says whether anybody should be told it exists.

@vs_agent_action(
    name="ag.update_settings",
    description="Update this agent's settings",
    http={"url": "/v1/agent/settings", "method": "PUT"},
    capability=False,
)
async def update_settings(request: SettingsRequest):
    ...

/capabilities and the MCP tool list answer "what can I give you to do". An agent's own management surface — settings, CRUD, anything a person reaches through a screen — still needs its route, but does not belong in that answer: advertising one puts it in front of a planner, which will eventually route a conversation into a CRUD call.

It defaults to True so that forgetting it leaves a capability visible, which is the safe direction to be wrong in.

Discover action modules and packages

Pass the action modules explicitly instead of maintaining an import for every file:

server = VsAgentServer(
    config=config,
    http="fastapi",
    mcp="fastmcp",
    authenticator=authenticator,
    action_modules=[
        "my_agent.actions",             # Recursively import this package
        "shared_actions.health_action", # Import only this module
    ],
)
server.add_actions()

Use absolute dotted Python import paths. A module entry imports that module; a package entry imports its initializer and recursively discovers its children with pkgutil. Explicit entries follow the supplied order, and discovered siblings are sorted by name. Overlapping entries are visited once per scan, and normal Python import caching prevents modules from executing again on subsequent scans. No reload is performed.

Imports happen during server construction, before transport servers are created. Import/discovery failures are logged with the module/package name and re-raised with context; duplicate-action and transport validation errors also abort construction. Keep action packages focused: all discovered modules execute their normal import-time code. This imports into the existing process-wide registry; it does not filter actions that were already registered.

add_actions() remains the explicit step that wires registered actions to HTTP/MCP. Omitting action_modules preserves manual imports.

Import order matters

@vs_agent_action self-registers at import time into VsActionRegistry. Import your action modules before calling server.add_actions().

import my_agent.actions.search   # noqa — triggers registration
import my_agent.actions.summary  # noqa

server.add_actions()
server.run()

---

@action — deprecated

@action is deprecated. Use @vs_agent_action. It still works and emits a DeprecationWarning.

Its shape bakes in "HTTP always, MCP whenever it is a capability", so an action cannot be HTTP-only without also disappearing from /capabilities, and cannot listen on a queue at all.

# before
@action(name="x", description="…", path="/v1/x", method="POST")

# after
@vs_agent_action(name="x", description="…", http={"url": "/v1/x", "method": "POST"}, mcp={})

@action parameter	@vs_agent_action equivalent
path, method	http={"url": path, "method": method}
capability=True	mcp={} plus capability=True
capability=False	omit mcp, pass capability=False
name, description, intents, input_schema, guards	unchanged

One behavioural note on migrating: @action registered its MCP tool at decoration time, which meant importing an action module required vs-mcp-agent to be installed. @vs_agent_action leaves that to add_actions(), since the server is the one making the decision.

---

Shared request authentication

With [agent] enable_auth = true, supply one authenticator instance for HTTP and MCP:

from vs_security import VsBearerTokenAuthenticator, VsJWTProvider, get_auth_context

# A custom authenticator can implement VsRequestAuthenticator.authenticate instead.
# For role guards, pass your VsAuthContext subclass to VsJWTProvider(context_class=...).
authenticator = VsBearerTokenAuthenticator(VsJWTProvider(config))
server = VsAgentServer(
    config, http="fastapi", mcp="fastmcp", authenticator=authenticator,
)
server.add_actions()

# Inside a handler or tool:
# context = get_auth_context()

authenticate(request: VsAuthRequest) -> VsAuthContext receives headers with case-insensitive lookup, a transport (http, mcp for MCP HTTP, or stdio), a path, and an HTTP method. Return validated identity or raise VsAuthenticationError; do not set the context yourself. See vs-security for the interface and bearer-session adapter. Authenticator implementations must support concurrent calls across transport threads and event loops.

The HTTP filter authenticates routes, controllers, actions, /health, and /capabilities; the existing public documentation paths remain excluded. WebSocket authentication continues to use WebSocket guards. MCP HTTP middleware authenticates each request except /.well-known/ metadata paths. Per-message MCP middleware rebinds the validated request identity inside session tasks. Both adapters reset context in finally, including failures and cancellation.

Authentication failures produce 401, insufficient permissions produce 403, and unexpected authentication errors are logged with tracebacks and propagated as server errors. When authentication is disabled, the supplied instance is unused; action guards still run. Peer discovery sends the existing [agent_client] service credentials when requesting health and capabilities.

---

MCP stdio authentication

Stdio uses the same VsRequestAuthenticator.authenticate() before each MCP request, followed by guards and the handler. Set [agent] transport = stdio and enable_auth = true; pass http=None for an MCP-only agent.

authenticator = VsBearerTokenAuthenticator(
    VsJWTProvider(config), stdio_token_env="VS_AGENT_TOKEN",
)
server = VsAgentServer(
    config, http=None, mcp="fastmcp", authenticator=authenticator,
)

The launcher supplies VS_AGENT_TOKEN in the server process environment. The built-in adapter validates that token through the same session used for HTTP bearer tokens, on every request. Missing or invalid credentials reject the request before the tool executes. A custom authenticator receives VsAuthRequest(transport="stdio", headers={}, method=<MCP method>) and can use its own trusted credential source. Environment credentials identify the launched process, not separate users within it. Stdio authentication errors are MCP protocol errors rather than HTTP status codes.

Configure VS logging to a file with console logging disabled before importing agent/security modules; stdout is reserved for MCP protocol messages. HTTP and MCP HTTP behavior remains as described above.

---

Authorization with VsActionSecurity

VsActionSecurity is a unified guard that works on both HTTP and MCP. It reads the single shared auth context populated through your VsRequestAuthenticator.

from vs_agent.auth.vs_action_security import VsActionSecurity

@vs_agent_action(
    name="search_docs",
    description="Search documentation",
    http={"url": "/v1/docs/search"},
    mcp={},
    guards=[VsActionSecurity(roles=["user"])],
)
async def search_docs(query: str) -> VsToolResponse:
    ...

No roles (auth only):

guards=[VsActionSecurity()]  # rejects unauthenticated callers, any role allowed

Multiple roles (any match):

guards=[VsActionSecurity(roles=["admin", "editor"])]  # passes if caller has admin OR editor

Multiple guards (all must pass, in order):

guards=[VsActionSecurity(roles=["admin"]), require_verified_account]

How guards run per protocol:

Protocol	Auth context source	Guard execution
HTTP	VsRequestAuthFilter calls the supplied authenticator and sets context	Guards run as route dependencies, in order, after the filter sets context
MCP	VsMcpAuthMiddleware authenticates; per-message middleware binds context before dispatch	Guards called in order before tool function executes
Queue	VsRequestAuthenticator authenticates message headers and queue metadata	Guards run before action execution and cached-result replay

---

Lifecycle Hooks

Use lifecycle hooks to run code at server startup and shutdown — creating DB tables, warming caches, closing connections, etc.

HTTP lifecycle (vs-server)

from vs_server.lifecycle.vs_lifecycle import startup, shutdown

@startup
async def warm_cache():
    await VsCacheManager.set("ready", True)

@shutdown
async def flush_cache():
    await VsCacheManager.delete("ready")

Register hook modules with server_registry:

from vs_server.decorator.vs_server_registry import server_registry

@server_registry(hooks="my_agent.lifecycle")
def main():
    ...

MCP lifecycle (vs-mcp-agent)

from vs_mcp_agent.lifecycle.vs_mcp_lifecycle import mcp_startup, mcp_shutdown

@mcp_startup
async def init_mcp_resources():
    await load_tool_index()

@mcp_shutdown
async def cleanup_mcp_resources():
    await close_tool_connections()

Register hook modules with mcp_server_registry:

from vs_mcp_agent.decorator.vs_mcp_server_registry import mcp_server_registry

@mcp_server_registry(hooks="my_agent.mcp_lifecycle")
def main():
    ...

When both are used, HTTP and MCP hooks run independently — HTTP hooks fire when the HTTP server starts/stops, MCP hooks fire when the MCP server starts/stops.

---

HTTP-only Capabilities

Declare http and omit mcp:

@vs_agent_action(
    name="ag.list_agents",
    description="List all agents",
    http={"url": "/v1/agents", "method": "GET"},
)
async def list_agents():
    ...

For plain routes that are not actions at all, use @controller from vs-server directly and register via server.add_controller():

from vs_server.decorator.vs_controller_decorator import controller, get, post

@controller("/v1/internal")
class InternalController:

    @get("/status")
    async def status(self):
        return {"status": "ok"}

server.add_controller(InternalController())

---

MCP-only Capabilities

Declare mcp and omit http:

@vs_agent_action(
    name="summarize",
    description="Summarize a document",
    mcp={"name": "summarize_document"},
)
async def summarize(request: SummarizeRequest) -> VsToolResponse:
    ...

For tools that are not actions at all, use @tool from vs-mcp-agent directly:

from vs_mcp_agent.decorator.tool import tool
from vs_mcp_agent.schema.vs_tool_response import VsToolResponse

@tool(name="internal_tool", description="Internal MCP tool only")
async def internal_tool(query: str) -> VsToolResponse:
    ...

These are picked up automatically by VsFastMcpServer at run() — no extra registration needed.

---

Queue request/reply transport

Install the queue extra and your chosen broker client:

pip install "vs-agent[queue]" "vs-queue[redis]"
# RabbitMQ: install "vs-queue[rabbitmq]" instead.

Pass a disconnected provider to the server. VsAgentServer.run() owns its connection, consumer tasks, and shutdown on one dedicated event loop. HTTP/MCP continue on their existing loops. Queue-only agents use http=None, mcp=None and run the queue loop in the main thread.

from vs_queue.factory.vs_queue_factory import VsQueueFactory

queue_client = VsQueueFactory.create(provider="redis", host="localhost", port=6379)

async def allow_reply(request, context):
    # reply_queue must come from validated identity, not from unverified payload data.
    if request.reply_to != context.metadata.get("reply_queue"):
        raise PermissionError("Reply destination is not permitted")

server = VsAgentServer(
    config=config,  # agent.enable_auth=true
    http="fastapi",
    mcp="fastmcp",
    authenticator=authenticator,
    queue=queue_client,
    queue_reply_guard=allow_reply,
    action_modules=["my_agent.actions"],
)
server.add_actions()
server.run()

An action opts into queue processing:

@vs_agent_action(
    name="calculator.add", description="Add two numbers",
    http={"url": "/v1/add", "method": "POST"},
    mcp={"name": "add"},
    queue={
        "listen": "calculator.requests",
        "reply_required": True,
        "max_retries": 3,
        "retry_backoff_seconds": 1.0,
        "dead_letter_queue": "calculator.dead",
    },
    guards=[require_calculator_access],
)
async def add(input: AddInput) -> AddOutput:
    return AddOutput(result=input.a + input.b)

A caller publishes through its own vs-queue connection:

from vs_queue.schema.vs_message import VsMessage

await caller_queue.publish("calculator.requests", VsMessage(
    headers={
        "Authorization": f"Bearer {access_token}",
        "reply_to": "caller.responses",
        "correlation_id": "request-123",
    },
    payload={"a": 10, "b": 20},
))

For a single Pydantic-model parameter, send the same flat body as HTTP: {"a": 10, "b": 20}. The legacy {"input": {"a": 10, "b": 20}} shape remains accepted when input is not a field name or validation alias in the model. Model fields take precedence if the shapes are ambiguous. For multiple parameters or primitive parameters, payload keys match function argument names, like MCP arguments. Pydantic annotations/defaults validate inputs and return values. Replies go to the per-message reply_to, never a constant action-level destination. Response payloads are {"status": "success", "result": ...} or {"status": "error", "error": {"code": ..., "status_code": ...}}. Only correlation_id and the action name are copied into response headers; authorization credentials are never copied. An absent correlation ID defaults to the request message ID.

The adapter authenticates with VsAuthRequest(transport="queue", topic=..., message_id=..., reply_to=..., correlation_id=...), authorizes the reply destination, validates input, executes guards, and invokes the action. Shared context is reset in finally, including cancellation. The reply guard accepts the request and validated context, and must raise or return False to reject a destination; successful guards return None or True. Authenticated queue configurations require this guard. Replies to any request queue owned by the agent are rejected to prevent loops.

Malformed headers, failed authentication, and unauthorized reply destinations are permanently rejected without publishing to an untrusted destination; they go to the configured dead-letter queue, or are discarded when none is configured. Invalid inputs and failed action guards publish safe 400/401/403 error responses to authorized reply destinations. Action execution and result-validation/serialization failures publish a generic internal_error/500 response before propagating to vs-queue retry handling and eventually dead-letter. These errors include error.attempt (one-based) and error.retryable (whether configured retries remain). A retryable error is an attempt notification: callers awaiting the final outcome should keep listening on the correlation ID for a success or an error with retryable=false. Failed attempts are not cached. If error publication also fails, that failure is logged and the original action exception still propagates. Callers should retain a response deadline for crashes, cancellation, unavailable brokers, or infrastructure failures that prevent replies. reply_required=False allows fire-and-forget messages without reply_to; authentication and action guards still apply.

The consumer publishes a response before returning; providers acknowledge only after successful processing or successful retry/dead-letter publication. Correlation IDs link responses to requests. Retries preserve the original request message ID. Completed results are cached before reply publication, so response-publish retries do not rerun the action within the cache window. Authentication and guards are rechecked before cached results are replayed. A message ID reused with different payload/headers produces a 409 duplicate conflict.

Cache settings are agent.queue_cache_size (default 1000 per action) and agent.queue_cache_ttl (default 3600 seconds). This cache is in memory and is not an exactly-once guarantee: process restarts, expiry, eviction, and failed actions can execute again. Side-effecting actions must implement durable idempotency in their application data store. from vs_agent.queue import get_queue_request exposes the current message ID, topic, and correlation ID to queue handlers/guards; it returns None outside a delivery and is reset on all exit paths. An application-level idempotency key in the input model can cover HTTP/MCP callers too. Queue startup/shutdown timeouts are agent.queue_startup_timeout (30 seconds) and agent.queue_shutdown_timeout (10 seconds). Register all actions before starting the runtime; one action owns each request queue.

Redis subscription restarts replay their own pending deliveries. Redis servers with XAUTOCLAIM support additionally reclaim stale deliveries from dead consumers. The provider option credentials={"claim_idle_ms": 300000} controls the idle threshold (default five minutes; zero disables reclaim). Set it longer than expected processing time, and retain application idempotency because an overdue delivery can be claimed while its original worker is still running. Older Redis servers log that automatic stale recovery is unavailable.

---

/capabilities Endpoint

server.add_actions() automatically registers a GET /capabilities endpoint on the HTTP server. It returns every action with capability=True — whatever transports it declares — so orchestrators can discover what the agent can do and how to dispatch it.

Actions with capability=False are excluded. They are still served on their routes; they are just not advertised.

Response:

{
  "actions": [
    {
      "name": "ag.create_new_agent",
      "description": "Create a new Viveka Sutra agent",
      "path": "/v1/agent",
      "method": "POST",
      "mcp_tool": "create_new_agent",
      "queue_topic": "agent.create",
      "available_transports": ["http", "mcp"],
      "transports": {
        "http": {
          "available": true,
          "path": "/v1/agent",
          "method": "POST",
          "url": "http://localhost:9011/v1/agent"
        },
        "mcp": {
          "available": true,
          "name": "create_new_agent",
          "description": "Create a new Viveka Sutra agent",
          "transport": "streamable-http",
          "url": "http://localhost:8080/mcp"
        },
        "queue": {
          "available": false,
          "topic": "agent.create",
          "reply_required": true,
          "reply_to_header": "reply_to",
          "correlation_id_header": "correlation_id"
        }
      },
      "intents": [],
      "input_schema": null
    }
  ]
}

available_transports lists adapters configured and registered for this action on this server. This is registration information, not a live health check. transports describes each declared transport; undeclared transports are null. HTTP paths include server.base_url, and the HTTP URL uses the discovery request's base URL. MCP details include its resolved tool name and description, configured transport, and the server's prepared connection URL. The MCP URL is null before preparation and for stdio. MCP wildcard bind addresses are displayed as localhost, as in the startup log; externally proxied deployments may require a different connection address.

Declaring MCP without configuring an MCP server reports available: false; a queue declaration reports available when it is registered with a supplied vs-queue instance. /capabilities is also registered when only MCP/queue actions are declared and an HTTP server is configured.

Queue details also advertise reply_required, reply_to_header, and correlation_id_header for dynamic request/reply routing.

The legacy path, method, mcp_tool, and queue_topic fields remain for compatibility and describe declarations, regardless of availability. They are null when that transport is undeclared.

---

Intents

Intent gives each action semantic labels that orchestrators use for routing — matching a user request to the right action without exact string matching.

from vs_agent.schema.vs_action_schema import Intent

@vs_agent_action(
    name="search_docs",
    description="Search VS library documentation",
    http={"url": "/v1/docs/search"},
    intents=[
        Intent(
            name="search",
            description="Find documentation matching a query",
            examples=["how do I configure logging", "what does VsLogManager do"],
        )
    ],
)
async def search_docs(query: str) -> VsToolResponse:
    ...

Intents are returned in /capabilities and are available as metadata on VsActionRegistry entries.

---

Error Handling

Exceptions from action functions propagate through whichever protocol the call arrived over:

• On HTTP, vs-server's exception handlers convert them to HTTP responses.
• On MCP, VsFastMcpServer returns an error result to the MCP client.

Use exceptions from vs-server for standard HTTP error semantics — they are handled automatically:

from vs_server.schema.exceptions import NotFoundException, ServiceUnavailableException

@vs_agent_action(name="get_doc", description="Get a document", http={"url": "/v1/docs/{doc_id}", "method": "GET"})
async def get_doc(doc_id: str) -> VsToolResponse:
    doc = await repo.find(doc_id)
    if doc is None:
        raise NotFoundException(f"Document '{doc_id}' not found")
    return VsToolResponse(status="success", result=doc)

---

Class Reference

---

VsAgentServer

Coordinates HTTP and MCP servers. Uses VsServerFactory and VsMcpServerFactory to resolve server implementations. Supports HTTP-only, MCP-only, or HTTP + MCP modes.

Constructor:

Parameter	Type	Default	Description
config	VsBaseConfig	—	Application config
http	Optional[str]	"fastapi"	HTTP server key registered in VsServerFactory. Pass None for MCP-only.
mcp	Optional[str]	None	MCP server key registered in VsMcpServerFactory. Pass "fastmcp" to enable MCP.
authenticator	Optional[VsRequestAuthenticator]	None	Shared authentication instance; required when agent.enable_auth=true.
action_modules	Optional[Sequence[str]]	None	Dotted action modules or packages to import; packages are scanned recursively.
queue	Optional[VsBaseQueue]	None	Disconnected vs-queue provider instance; connected and managed by the queue runtime
queue_reply_guard	Callable	None	Authorizes (VsAuthRequest, AuthContext) reply destinations; required for authenticated queue agents

Methods:

Method	Signature	Description
add_controller	add_controller(controller) -> VsAgentServer	Register an HTTP controller. Requires http mode.
add_router	add_router(router) -> VsAgentServer	Register a raw router. Requires http mode.
add_websocket	add_websocket(handler) -> VsAgentServer	Register a WebSocket handler. Requires http mode.
add_sse	add_sse(handler) -> VsAgentServer	Register an SSE handler. Requires http mode.
add_actions	add_actions() -> VsAgentServer	Wire every registered action to the transports it declares: HTTP routes and /capabilities, and MCP tools via VsToolRegistry.
get_app	get_app() -> Any	Return the underlying ASGI app. Requires http mode.
run	run() -> None	Start all servers. MCP on daemon thread, HTTP on main thread.

Property:

Property	Type	Description
mcp	Any	The underlying FastMCP instance. Raises RuntimeError if MCP not configured.

Notes:

• All add_* methods call _require_http() internally and raise RuntimeError if http=None.
• In HTTP + MCP mode, run() starts MCP on a background daemon thread then blocks on the HTTP server.
• Import VsFastApiServer before constructing VsAgentServer to ensure "fastapi" is registered.
• Import VsFastMcpServer before constructing VsAgentServer to ensure "fastmcp" is registered.
• add_actions() logs a warning for actions declaring a transport whose server is not configured, including queue actions without a configured queue instance.

---

@vs_agent_action

Decorator. Registers a function in VsActionRegistry with the transports it declares. Self-registers at import time.

Parameters:

Parameter	Type	Required	Default	Description
name	str	Yes	—	Action key. Must be unique
description	str	Yes	—	Shown in /capabilities and the MCP tool list
http	Optional[HttpTransport | dict]	No	None	Present → HTTP route
mcp	Optional[McpTransport | dict]	No	None	Present → MCP tool
queue	Optional[QueueTransport | dict]	No	None	Present → queue consumer when a vs-queue instance is configured
intents	List[Intent]	No	[]	Semantic intents for orchestrator routing
input_schema	Dict[str, Any]	No	None	JSON schema for the action input
guards	List[Callable]	No	[]	Applied on every transport, in order
capability	bool	No	True	Whether the action is advertised

Notes:

• Each name must be unique across all registrations. Duplicates raise ValueError.
• An action with no transport raises ValueError.
• guards must be async callables.
• Transport dicts are validated into their pydantic models, so a bad key fails at import time rather than at request time.

---

HttpTransport / McpTransport / QueueTransport

Pydantic models for the transport blocks. Dicts passed to @vs_agent_action are validated into these.

HttpTransport:

Field	Type	Default	Description
url	str	—	Endpoint path. Also accepted as path
method	str	"POST"	HTTP method

McpTransport:

Field	Type	Default	Description
name	Optional[str]	None → the action name	MCP tool name
description	Optional[str]	None → the action description	MCP tool description

QueueTransport:

Field	Type	Default	Description
listen	str	—	Queue topic to consume

---

@action (deprecated)

Decorator. Kept working for existing agents; emits a DeprecationWarning. Delegates to @vs_agent_action as http={"url": path, "method": method} plus mcp={} when capability is true.

Parameters:

Parameter	Type	Required	Default	Description
name	str	Yes	—	MCP tool name and action key
description	str	Yes	—	Shown in MCP tool list and /capabilities
path	str	Yes	—	HTTP endpoint path
method	str	No	"POST"	HTTP method
intents	List[Intent]	No	[]	Semantic intents for orchestrator routing
input_schema	Dict[str, Any]	No	None	JSON schema for the action input
guards	List[Callable]	No	[]	Applied on both HTTP and MCP, in order
capability	bool	No	True	Whether the action is advertised and exposed over MCP

---

VsActionSecurity

Guard class for unified HTTP + MCP authorization. Reads from the auth context set by whichever protocol is active.

Constructor:

Parameter	Type	Default	Description
roles	Optional[List[str]]	[]	Required roles. Caller must have at least one. Empty list = any authenticated caller.

Behaviour:

Condition	Result
No auth context	Raises VsAuthenticationError (401)
Auth context present, no roles required	Passes
Auth context present, caller has required role	Passes
Auth context present, caller lacks required role	Raises VsInsufficientRolesError (403)

Notes:

• On HTTP, VsRequestAuthFilter calls the supplied authenticator and sets the shared context.
• On MCP, auth context is set by VsMcpAuthMiddleware (from vs-mcp-agent).
• VsActionSecurity reads from the same context variable regardless of protocol.

---

VsActionRegistry

Class-level registry of all registered actions. Thread-safe.

Methods:

Method	Signature	Description
register_action	register_action(name, description, intents, input_schema, guards, fn, capability=True, http=None, mcp=None, queue=None) -> None	Register an action. Raises ValueError if the name is taken
register	register(name, description, path, method, intents, input_schema, guards, fn, capability=True) -> None	Deprecated. The @action shape; delegates to register_action
get_all	get_all() -> Dict[str, _ActionEntry]	Every action, whatever it is reachable over
capabilities	capabilities() -> Dict[str, _ActionEntry]	Only actions with capability=True
http_actions	http_actions() -> Dict[str, _ActionEntry]	Only actions declaring an http transport
mcp_actions	mcp_actions() -> Dict[str, _ActionEntry]	Only actions declaring an mcp transport
queue_actions	queue_actions() -> Dict[str, _ActionEntry]	Only actions declaring a queue transport

_ActionEntry fields: name, description, intents, input_schema, guards, fn, capability, http, mcp, queue.

_ActionEntry derived properties: path and method (from http, None without it), mcp_name and mcp_description (from mcp, falling back to the action's own name and description).

---

Intent

Pydantic model. Semantic label for an action — used by orchestrators for routing.

Fields:

Field	Type	Required	Description
name	str	Yes	Short intent identifier
description	str	Yes	What this intent means
examples	Optional[List[str]]	No	Example user phrases that trigger this intent

---

ActionCapability

Pydantic model. Metadata for a single action as returned by /capabilities.

Fields:

Field	Type	Description
name	str	Action name
description	str	Action description
path	Optional[str]	HTTP endpoint path. None without an http transport
method	Optional[str]	HTTP method. None without an http transport
mcp_tool	Optional[str]	MCP tool name. None without an mcp transport
queue_topic	Optional[str]	Queue topic. None without a queue transport
available_transports	List[str]	Adapters configured and registered for this action
transports	ActionTransports	HTTP/MCP/queue details, including availability, endpoint information, and MCP description
intents	List[Intent]	Semantic intents
input_schema	Optional[Dict[str, Any]]	JSON schema for input

---

CapabilitiesResponse

Pydantic model. Response from GET /capabilities.

Fields:

Field	Type	Description
actions	List[ActionCapability]	All advertised actions — those with capability=True