vs-agent

Full-stack agent framework for Viveka Sutra — build agents that serve over HTTP, MCP, or both, 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 expose a capability once with @action and have it available simultaneously as an HTTP endpoint and an MCP tool — with the same auth, guards, and business logic.

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

@action(
    name="search_docs",
    description="Search VS library documentation",
    path="/v1/docs/search",
    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.

---

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

---

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.auth_enabled	false	Enable JWT auth middleware on MCP

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
auth_enabled = 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 @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.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  # @action decorators self-register into VsActionRegistry + VsToolRegistry

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

server.add_actions()
    ├── reads VsActionRegistry → wires HTTP routes + /capabilities
    └── VsFastMcpServer already has tools from VsToolRegistry (wired at run())

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.

At least one of http or mcp must be specified — both None 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 all @action functions to HTTP routes and expose /capabilities.
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()

---

@action

Registers a function simultaneously as:

• An MCP tool — in VsToolRegistry, picked up by VsFastMcpServer when run() is called
• An HTTP endpoint — in VsActionRegistry, wired by server.add_actions()

from vs_agent.decorator.vs_action_decorator import action
from vs_mcp_agent.schema.vs_tool_response import VsToolResponse

@action(
    name="search_docs",
    description="Search VS library documentation",
    path="/v1/docs/search",
    method="POST",
    guards=[VsActionSecurity(roles=["user"])],
)
async def search_docs(query: str) -> VsToolResponse:
    results = await _do_search(query)
    return VsToolResponse(status="success", result=results, summary="Search complete")

Parameters:

Parameter	Type	Required	Default	Description
name	str	Yes	—	MCP tool name and action identifier
description	str	Yes	—	Shown in MCP tool list and /capabilities
path	str	Yes	—	HTTP endpoint path
method	str	No	"POST"	HTTP method (GET, POST, PUT, DELETE, PATCH)
intents	List[Intent]	No	[]	Semantic intents for agent routing
input_schema	Dict[str, Any]	No	None	JSON schema for the action input
guards	List[Callable]	No	[]	Guards applied on both HTTP and MCP

Import order matters. @action self-registers at import time into both VsActionRegistry and VsToolRegistry. Import your action modules before calling server.add_actions() or server.run().

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

server.add_actions()
server.run()

---

Authorization with VsActionSecurity

VsActionSecurity is a unified guard that works on both HTTP and MCP. It reads the auth context set by VsSecurityFactory (HTTP) or VsMcpAuthMiddleware (MCP).

from vs_agent.auth.vs_action_security import VsActionSecurity

@action(
    name="search_docs",
    description="Search documentation",
    path="/v1/docs/search",
    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	VsSecurityFactory.get() sets context via FastAPI Depends	Guards called in order after auth context is set
MCP	VsMcpAuthMiddleware sets context before tool dispatch	Guards called in order before tool function executes

---

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

For capabilities that should only be available over HTTP, 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

For capabilities that should only be available over MCP, 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.

---

/capabilities Endpoint

server.add_actions() automatically registers a GET /capabilities endpoint on the HTTP server. It returns all registered actions with their metadata — used by orchestrators to discover what the agent can do.

Response:

{
  "actions": [
    {
      "name": "search_docs",
      "description": "Search VS library documentation",
      "path": "/v1/docs/search",
      "method": "POST",
      "intents": [],
      "input_schema": null
    }
  ]
}

---

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

@action(
    name="search_docs",
    description="Search VS library documentation",
    path="/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 both protocols:

• 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

@action(name="get_doc", description="Get a document", path="/v1/docs/{doc_id}")
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.

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 all @action functions to HTTP routes and /capabilities.
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.

---

@action

Decorator. Registers a function as both an MCP tool (via VsToolRegistry) and an HTTP action (via VsActionRegistry). Self-registers at import time.

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

Notes:

• Each name must be unique across all @action registrations. Duplicate names raise ValueError.
• method is HTTP-only — MCP tools have no HTTP method concept.
• guards must be async callables.

---

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 PermissionError("Unauthenticated request")
Auth context present, no roles required	Passes
Auth context present, caller has required role	Passes
Auth context present, caller lacks required role	Raises PermissionError

Notes:

• On HTTP, auth context is set by VsSecurityFactory.get() (from vs-security).
• 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 @action-registered functions. Thread-safe.

Methods:

Method	Signature	Description
register	register(name, description, path, method, intents, input_schema, guards, fn) -> None	Register an action. Raises ValueError if name is already registered.
get_all	get_all() -> Dict[str, _ActionEntry]	Returns a snapshot of all registered actions.

Notes:

• @action calls VsActionRegistry.register() and VsToolRegistry.register_fn() at decoration time.
• server.add_actions() reads from VsActionRegistry.get_all() to wire HTTP routes.

---

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	str	HTTP endpoint path
method	str	HTTP method
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 registered actions