NOTICE: All information contained herein is, and remains
the property of TechnoCore.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore and dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore.
Base Axion MCP (Model Context Protocol) server. Wraps the FastMCP SDK
and provides a registration point for Axion tool groups so that external
AI clients can discover and invoke Axion capabilities.
Note: This is entirely separate from
factory.ai/package.mcp/,
which is the Multi-Cloud Provider abstraction for LLM backends.
ObjMcpServermakes Axion act as an MCP server;ObjMcpClient
makes Axion act as an MCP client connecting to external servers.
AI Client (Claude Desktop / Cursor / Ollama)
│ MCP protocol (stdio or SSE)
▼
ServeMcp.py
│
▼
ObjMcpServer.build_app()
│ registers tool groups
▼
FastMCP app
│ calls
▼
Axion objects (ObjData, ObjWorkflow, ObjNotify, ObjAI …)
Each tool group is registered by a _register_*_tools(app) method.
Later phases add groups by adding new register methods and calling them
inside build_app().
server = ObjMcpServer(db=0)
db: Database connection instance (default 0 = auto-connect).build_app(name="axion") -> FastMCPCreate and return a configured FastMCP application with all
registered tool groups.
mcp_app = server.build_app(name="axion")
mcp_app.run(transport="stdio")
Raises ImportError when the mcp package is not installed.
health_check() -> dictCheck database reachability and return the active package name.
{"status": "ok", "package": "homechoice"}
get_package() -> strReturn the active Axion package name (AXION_PACKAGE env var →
package: key in config.yaml → "base").
get_config_value(section, key) -> strRead a single value from the config.yaml hierarchy with active-package
override resolution.
get_config_value("database", "host") → "localhost"
get_config_value("mqtt", "port") → "1883"
list_ai_providers() -> list[str]Scan factory.ai/package.mcp/ and return sorted provider names that
can be used in ObjAI model strings.
["anthropic", "database", "google", "huggingface",
"mistral", "ollama", "openai", "playwright"]
ServeMcp.py in the project root is the service entry point.
python ServeMcp.py serve
python ServeMcp.py serve --transport sse --host 0.0.0.0 --port 8080
Add to ~/.config/claude/claude_desktop_config.json:
{
"mcpServers": {
"axion": {
"command": "python",
"args": ["/path/to/axion/ServeMcp.py", "serve"]
}
}
}
query_sql(sql, limit=100) -> list[dict]Execute a read-only SQL query and return rows as a list of dicts.
Only SELECT, SHOW, DESCRIBE, EXPLAIN, and WITH are accepted.
Any statement containing write keywords (INSERT, UPDATE, DELETE,
DROP, TRUNCATE, …) is rejected. A LIMIT is appended automatically
when missing, capped at 500 rows.
query_sql("SELECT TableName, Active FROM def_workflow LIMIT 10")
get_schema(table_name) -> strReturn the CREATE TABLE DDL for a table using SHOW CREATE TABLE.
Returns "" when the table does not exist.
get_schema("def_workflow")
list_tables(filter="") -> list[str]List table names in the active database. filter accepts a SQL LIKE
pattern (e.g. "def_%", "bloom_%"). Returns a sorted list of strings.
list_tables("def_%")
list_tables() # all tables
list_workflows(filter="", active_only=True) -> list[dict]List workflows from def_workflow for the active package. filter
is a substring match on WorkflowName. Returns WorkflowName, Package,
Description, Async, Active, ThreeWords.
workflow_status(workflow_code) -> dictReturn metadata for a single workflow by its exact WorkflowName.
trigger_workflow(workflow_code, context={}) -> dictExecute a workflow via Workflow.Run(). Async workflows are queued
immediately ("async": true, empty result). Sync workflows execute
inline and return their result context.
trigger_workflow("SCORE_APPLICATION", {"param1": "ID123"})
Returns {workflow_code, guid, async, result}.
list_calculations(filter="") -> list[dict]List calculation groups from def_calculation. filter is a substring
match on CalculationGroup.
run_calculation(group_name, guid="", context={}) -> dictExecute a calculation group via ObjCalculation.run_calculation_set().
Steps run in Rank order; CALC_FAIL notifications are sent on step errors.
run_calculation("CREDIT_SCORE", context={"param1": "12345"})
Returns {group_name, guid, result}.
list_channels(filter="") -> list[dict]List notification codes from def_notify for the active package.
Returns NotifyCode, Description, NotifyMode, Active. filter is a
substring match on NotifyCode.
send_notification(notify_code, message="") -> dictTrigger a notification by its code. The delivery channel (Slack,
Discord, SMS, MQTT, Email, …) and recipients are configured in
def_notify. Lazy-imports ObjNotify with all required paths.
send_notification("SYSADMIN", "Deployment complete.")
list_models() -> dictReturns {providers, ollama_available, ollama_models}.
providers lists factory.ai/package.mcp/ names; ollama_models
lists installed Ollama model tags when the daemon is reachable.
ai_prompt(model, role, prompt) -> strSend a prompt to any configured AI model via ObjAI.
ai_prompt("mcp:ollama:mistral",
"You are a SQL expert.",
"Explain this query: SELECT * FROM def_workflow")
ai_prompt("mcp:anthropic:claude-opus-4-6",
"You are a data analyst.",
"Summarise this dataset…")
ai_embed(text, model="nomic-embed-text") -> list[float]Generate an embedding vector via a local Ollama embedding model.
Returns [] when Ollama is unavailable.
ai_embed("invoice processing workflow", "mxbai-embed-large")
| Phase | Register method | Tools added |
|---|---|---|
| 5 | _register_doc_tools |
list_documents, get_document |
Each phase adds a _register_*_tools(app) method to ObjMcpServer
and a single call to it inside build_app().
| Test Suite | Tests | Status | Purpose |
|---|---|---|---|
test_ObjMcpServer.py |
71 unit + 2 integration | ✅ All passed | Full tool surface |
dev-env/bin/pytest resource.test/pytests/factory.core/test_ObjMcpServer.py -v
TestIsReadOnlySql (25 parametrized cases) — _is_read_only_sql() guards the
query_sql tool against write operations.
| Category | Examples | Expected |
|---|---|---|
| Read-only | SELECT, SHOW, DESCRIBE, EXPLAIN, WITH … SELECT |
True |
SHOW CREATE TABLE |
Regression case — was incorrectly blocked | True |
| Write | INSERT, UPDATE, DELETE, DROP, ALTER, TRUNCATE, REPLACE, EXEC |
False |
| Semicolon injection | SELECT … ; DROP TABLE foo |
False |
Empty / blank / only ; |
False |
TestSerializeDict (7 tests) — _serialize_dict() converts database types to
JSON-safe values before returning rows to the MCP client.
datetime → ISO-8601 string ("2024-01-15T12:30:00")date → ISO date string ("2024-06-01")Decimal → string representation ("12.50")bytes → decoded UTF-8 stringNone → preserved as Nonestr, int, float → unchangeddict → recursedTestSerializeRows (4 tests) — _serialize_rows() applies _serialize_dict
across a list of row dicts, preserving row count and non-typed fields.
TestBuildApp (4 tests) — build_app() construction contract.
FastMCP instanceEXPECTED_TOOLS set)ImportError with a clear message when the mcp package is missingTestCoreTools (6 tests) — Phase 1 tools.
| Tool | What is tested |
|---|---|
health_check |
Returns {"status":"ok","package":…} on success; {"status":"error","detail":…} on DB error |
get_package |
Returns active package string |
get_config_value |
Delegates to get_ini_value(section, key); returns "" on KeyError |
list_ai_providers |
Returns [] on RuntimeError |
TestDataTools (9 tests) — Phase 2 tools.
| Tool | What is tested |
|---|---|
query_sql |
Rejects write SQL; auto-appends LIMIT; does not double-add LIMIT; serializes datetime fields in rows |
get_schema |
Returns DDL string; returns "" for empty name or missing table |
list_tables |
Returns list from SQL; returns [] on error |
TestWorkflowTools (4 tests) — Phase 3 workflow tools.
| Tool | What is tested |
|---|---|
list_workflows |
Returns serialized rows from def_workflow |
workflow_status |
Returns {"error": "…"} when workflow code not found |
trigger_workflow |
Sync execution returns {workflow_code, guid, async:False, result}; error path returns {"error":…, "workflow_code":…} |
TestCalcTools (3 tests) — Phase 3 calculation tools.
| Tool | What is tested |
|---|---|
list_calculations |
Returns rows with CalculationGroup |
run_calculation |
Returns {group_name, guid, result}; returns {"error":…} on failure |
TestNotifyTools (3 tests) — Phase 4 notification tools.
| Tool | What is tested |
|---|---|
list_channels |
Returns rows with NotifyCode |
send_notification |
Calls ObjNotify.Run(code, message_text=…) and returns {"sent":True}; returns {"error":…} on failure |
TestAITools (4 tests) — Phase 4 AI tools.
| Tool | What is tested |
|---|---|
list_models |
Returns {providers, ollama_available, ollama_models} structure |
ai_prompt |
Returns string from ObjAI.prompt(); returns "" on error |
ai_embed |
Returns [] on error |
TestIntegration (2 tests, @pytest.mark.integration) — require a live database.
health_check returns {"status":"ok"} with a real package stringlist_tables returns a non-empty list