feat: add python source config model
This commit is contained in:
@@ -7,6 +7,7 @@ from .models import (
|
||||
HttpSourceTransportConfig,
|
||||
LocalTargetConfig,
|
||||
McpSourceConfig,
|
||||
PythonSourceConfig,
|
||||
RpcHttpTargetConfig,
|
||||
RpcHttpTransportConfig,
|
||||
ServerConfig,
|
||||
@@ -25,6 +26,7 @@ __all__ = [
|
||||
"HttpSourceTransportConfig",
|
||||
"LocalTargetConfig",
|
||||
"McpSourceConfig",
|
||||
"PythonSourceConfig",
|
||||
"RpcHttpTargetConfig",
|
||||
"RpcHttpTransportConfig",
|
||||
"ServerConfig",
|
||||
|
||||
+23
-1
@@ -98,6 +98,28 @@ class StdlibSourceConfig(WorkflowConfigModel):
|
||||
id: Literal["wf.std", "wf.recipes"]
|
||||
|
||||
|
||||
class PythonSourceConfig(WorkflowConfigModel):
|
||||
"""Static config for trusted in-process Python workflow sources."""
|
||||
|
||||
kind: Literal["python"] = "python"
|
||||
id: str
|
||||
enabled: bool = True
|
||||
module: str = Field(min_length=1)
|
||||
registry: str = Field(default="registry", min_length=1)
|
||||
|
||||
@field_validator("id")
|
||||
@classmethod
|
||||
def validate_source_id(cls, value: str) -> str:
|
||||
if not re.fullmatch(SOURCE_ID_PATTERN, value):
|
||||
raise ValueError(
|
||||
"source id must start with alphanumeric or underscore and contain "
|
||||
"only [A-Za-z0-9_.-]"
|
||||
)
|
||||
if "." not in value:
|
||||
raise ValueError("source id must look like '<namespace>.<name>'")
|
||||
return value
|
||||
|
||||
|
||||
class McpSourceConfig(WorkflowConfigModel):
|
||||
"""Neutral config shape for MCP-backed workflow capability sources.
|
||||
|
||||
@@ -134,7 +156,7 @@ class McpSourceConfig(WorkflowConfigModel):
|
||||
|
||||
|
||||
SourceConfig = Annotated[
|
||||
StdlibSourceConfig | McpSourceConfig,
|
||||
StdlibSourceConfig | PythonSourceConfig | McpSourceConfig,
|
||||
Field(discriminator="kind"),
|
||||
]
|
||||
|
||||
|
||||
@@ -386,6 +386,30 @@ def test_load_workflow_config_resolves_role_store_paths_relative_to_config(
|
||||
)
|
||||
|
||||
|
||||
def test_workflow_config_parses_python_source() -> None:
|
||||
config = WorkflowConfigFile.model_validate(
|
||||
{
|
||||
"version": 1,
|
||||
"server": {
|
||||
"sources": [
|
||||
{
|
||||
"kind": "python",
|
||||
"id": "local.ops",
|
||||
"module": "tests.fixtures.python_source_ops",
|
||||
"registry": "registry",
|
||||
}
|
||||
]
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
source = config.server.sources[0]
|
||||
assert source.kind == "python"
|
||||
assert source.id == "local.ops"
|
||||
assert source.module == "tests.fixtures.python_source_ops"
|
||||
assert source.registry == "registry"
|
||||
|
||||
|
||||
def test_server_config_resolves_missing_role_stores_to_default_store() -> None:
|
||||
config = WorkflowConfigFile.model_validate(
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user